java - 如何在Hadoop Mapreduce中比较来自同一文件的数据？

Question

我有一个数据集看起来像这样:

X, Y, Text  
52.2552455,-7.5450262,donec
57.6727414,-4.269928,nulla
13.0504833,50.3308509,curae
25.6538807,58.263232,magna
30.292001,57.8308498,massa
37.9273983,41.107107,mauris
37.444498,2.9372148,volutpat   

X,Y值(经度，纬度)表示一个“点”。
我在这里想要实现的是通过每个x，y(作为一个值-键)之间的距离进行分组比较。
输出应为:

 52.2552455,-7.5450262 [nulla,curae,mauris]
 57.6727414,-4.269928, [curae,massa]
 13.0504833,50.3308509, [massa]
 25.6538807,58.263232, [volutpat,magna,mauris]
 30.292001,57.8308498, [mauris]
 37.9273983,41.107107, [mauris,volutpat,magna,curae]
 37.444498,2.9372148, [volutpat]

输出表示:52.2552455，-7.5450262与之相邻。
字符串(nulla，curae，mauris)。
数据集中所有x，y相同。

我已经写了下面的代码:

public static class Map extends Mapper<LongWritable, Text, Text, Text>
{           
    double spotX = 48.5672; 
    double spotY = 35.6897;
    String tweet = "";
    private int distX= 20;
    private int distY= 20;              
    private Text key1 = new Text();
    private Text value1 = new Text();


    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException 
        {   

            String line = value.toString();
            String[] results = line.split(",");

            for(int i=0; i < results.length; i+=3)
                {   

                    if (Double.parseDouble(results[i]) - spotX >= distX || Double.parseDouble(results[i+1]) - spotY >= distY )
                    {

                        key1.set(spotX + "," + spotY);
                        value1.set(results[i] + "," + results[i+1]+ "," +results[i+2]);
                        context.write(key1,value1); 
                    }
                    else
                    {
                        break;
                    }

                }
        }           
    }


public static class Reduce extends Reducer<Text, Text, Text, Text>
{

     public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException 
     {   

        List<String> vArrayList = new ArrayList<String>();
        for(Text v : values)
        {
            vArrayList.add(v.toString());
        }
        context.write(key, new Text(vArrayList.toString()));             
     }
}

现在，我的代码只能将所有“ Blob ”与map函数中声明的静态spotX(SpotY)进行比较。
有什么办法可以动态方式设置SpotX，SpotY？我的意思是说，将每一行都声明为可以与其他行进行比较的行吗？
我希望我足够清楚。

Answer 1

为了使您想要的成为可能，必须立即读取整个文件。您不能使用在输入文件的换行符上分割的默认TextInputFormat。

您可以在此处找到WholeFileInputFormat的实现。

http://bigdatathinker.blogspot.com/2014/01/reading-complete-file-in-mapreduce.html

您将在main方法的Job配置中使用它。看起来该链接的输入格式的键是Text，但值是BytesWritable。您可以找到其他示例，我确定可以读取整个文件。

您希望在映射器中得到的结果将收到"X, Y, Text\n52.2552455,-7.5450262,donec
\n57.6727414,-4.269928,nulla"(一个长字符串)，例如作为要处理的数据。

从那里，您可以value.split("\\n")，并遍历直线，并在拥有所有数据点时相应地进行距离计算。

总而言之，您在这里使用Hadoop并没有任何好处，因为1)您的数据集看起来很小2)它的性能仅比一个Java进程自己读取文件要好。

您可能需要研究的是K-means聚类，因为我认为这就是您似乎想要的。