java - 在JavaRDD中使用Text数据类型并在FlatMap中返回void

Question

我正在尝试将hadoop代码迁移到spark中。我已经有了一些预定义的函数，它们应该可以在spark中重用，因为它们只是Java代码，没有太多的hadoop依赖性。我有一个函数，可以接受文本格式的输入(空间数据经度，纬度)，并将其转换为形状(多边形，线流等)。当我尝试在Spark中读取它时，我首先以String形式读取文件的每一行。然后将它们转换为Text，以便可以使用以前创建的函数。但是我有两个疑问，首先，似乎JavaRDD不使用Text，对此我遇到了一些问题。其次，将Text转换为shape的函数不返回任何内容。但是我无法使用flatMap或任何其他映射技术。我什至不确定我的方法是否正确。

这是我的代码模型:

/*function for converting Text to Shape*/   
public interface TextSerializable {
public Text toText(Text text);
public void fromText(Text text);
* Retrieve information from the given text.
* @param text The text to parse
*/
}



/*Shape Class looks something like this*/

public interface Shape extends Writable, Cloneable, TextSerializable {
/
* Returns minimum bounding rectangle for this shape.
* @return The minimum bounding rectangle for this shape
*/
public Rectangle getMBR();

/**
* Gets the distance of this shape to the given point.
* @param x The x-coordinate of the point to compute the distance to
* @param y The y-coordinate of the point to compute the distance to
* @return The Euclidean distance between this object and the given point
*/
......
......
......*/

/*My code structure*/

 SparkConf conf = new SparkConf().setAppName("XYZ").setMaster("local");
 JavaSparkContext sc =new JavaSparkContext(conf);

 final Text text=new Text();

 JavaRDD<String> lines = sc.textFile("ABC.csv");

 lines.foreach(new VoidFunction<String>(){
 public void call(String lines){
        text.set(lines);
        System.out.println(text);
    }
    });

/*Problem*/
text.flatMap(new FlatMapFunction<Text>(){
    public Iterable<Shape> call(Shape s){
        s.fromText(text);
        //return void;
    }

代码的最后一行是错误的，但我不知道如何解决。 JavaRDD可以与用户定义的类一起使用(据我所知)。我什至不确定我是否已经将String行转换为Text文本的方式(如果RDD中允许这样做)。我是Spark的新手。任何帮助都将是巨大的。

Answer 1

您完全不了解这个概念。首先，您无法在任何对象上调用诸如map，flatmap等功能，它们只能从JavaRDD调用，而Text不是JavaRDD，Spark确实支持Text，但不支持您使用它的方式。

现在要问您的问题，因为您要将字符串转换为文本格式，请使用类似以下的内容

   SparkConf conf = new SparkConf().setAppName("Name of Application");
    JavaSparkContext sc = new JavaSparkContext(conf);
    JavaRDD<String> logData = sc.textFile("replace with address of file");

/*This map function will take string as input because we are calling it on javaRDD logData and that logData return string type value. This map fucntion will give Text as output 
you can replace the return statement with logic of your toText function(However new Text(s) is also a way to convert string into Text) but remember use of return is mandatory so apply logic accordingly 
         */
        JavaRDD<Text> rddone = logData.map(new Function<String,Text>(){
            public Text call(String s)
            {// type logic of your toText() function here
             return  new Text(s);}});  

现在，当我们通过JavaRDD rddone调用flatmap函数时，它将输入为文本，因为rddone的输出是Text，并且它可以根据需要提供输出。

/* This flatmap fucntion will take Text as input and will give iterator over object */
    JavaRDD <Object> empty = rddone.flatMap(new FlatMapFunction<Text,Object>(){
            public Iterator<Object> call(Text te)
            {
                // here you can call your fromText(te) method.
                return null;
        }
        }); 

另请参阅这些链接以获取更多详细信息 http://spark.apache.org/docs/latest/programming-guide.html

http://spark.apache.org/docs/latest/api/java/index.html?org/apache/spark/api/java/JavaRDD.html