如何使用Spark处理CSV数据并输出序列化的自定义JavaObjects？

Question

public class Cat{
    private Date dateOfAdoption;
    private String name;
    private Boolean isMale;
    private Person owner;

    public Cat(String name, Boolean isMale, Person owner, Date dateOfAdoption){
       this.name = name;
       this.isMale = isMale;
       this.owner = owner;
       this.dateOfAdoption = dateOfAdoption;
    }
}

public class AdoptedCats{
    public List<Cat> catList;

    public AdoptedCats(){
    }
}

AdoptedCats adoptedCats = <the adoptedcatjavaobjectbinaryfile>
System.out.println(adoptedCats.cats.length) // this would output say, 75 if there were 75 rows in the original CSV.

我想我必须使用Java Spark...但我不确定如何创建spark脚本，以便它可以分治整个一月的所有cat目录，并将它们输出到各自的目录中。任何帮助或例子都是最有帮助的。

Answer 1

实现这一目标的步骤-

>

假设文件（我有一个包含目录的小CSV文件的文件夹）是/somepath/catmydata.CSV

    JavaSparkContext context  = ......;

    //Retrieve list of files from master file

    List<String> files = context.textFile("/somepath/catmydata.csv")
            .flatMap(row->Arrays.asList(row.split(",")).iterator()).collect();

   //Iterate through each file and save it as object file
   Map<String,JavaRDD<Cat>> adoptedCatsMap =new HashMap<String,JavaRDD<Cat>>();
    for (String file:files) {

       // Extract file path and attach to convert
        String output=String.format("/convertedcatdata/%s",file.substring(file.indexOf('/'),file.length()))

       JavaRDD<Cat> adoptedCatsRdd = context.textFile(file).map(s->new Cat(s));

       adoptedCatsMap.put(output,adoptedCatsRdd);
       //This will write number of files based on rdd partitions,
       // if you want only one file in a directory then save it 
       // to temporary directory then use copyMerge to create one file using Hadoop fs FileUtil
       // https://hadoop.apache.org/docs/r2.7.1/api/org/apache/hadoop/fs/FileUtil.html
      //Save to physical location if required.
       adoptedCatsRdd.saveAsObjectFile(output);
    }

我希望这能有所帮助。