在 Spark 中将数据转换为 Parquet

Question

我在S3有一些遗留数据，我想使用Spark 2和Java API将它们转换成parquet格式。

我有所需的Avro模式（. avsc文件）及其使用Avro编译器生成的Java类，我想使用这些模式以Parque格式存储数据。输入数据不是任何标准格式，但我有一个库，可以将遗留文件中的每一行转换为Avro类。

是否可以将数据作为 JavaRDD 读取

类似于:

JavaRDD<String> rdd = javaSparkContext.textFile("s3://bucket/path_to_legacy_files");    
JavaRDD<MyAvroClass> converted = rdd.map(line -> customLib.convertToAvro(line));    
converted.saveAsParquet("s3://bucket/destination"); //how do I do this

像上面这样的东西可行吗？我后来想使用Hive，Presto和Spark处理转换后的镶木地板数据。




Answer 1

基本上是Steve提到的方法，但Hadoop输入和输出格式已经存在：

         Job job = new Job();
         ParquetOutputFormat.setWriteSupportClass(job, AvroWriteSupport.class);
         AvroParquetOutputFormat.setSchema(job, MyAvroType.SCHEMA$);
         AvroParquetOutputFormat.setBlockSize(job, 128*1024*1024);
         AvroParquetOutputFormat.setCompression(job, CompressionCodecName.SNAPPY);
         AvroParquetOutputFormat.setCompressOutput(job, true);

         sparkContext.textFile("s3://bucket/path_to_legacy_files")
            .map(line -> customLib.convertToAvro(line))
            .mapToPair(record -> new Tuple2<Void, MyAvroType>(null, record))
            .saveAsNewAPIHadoopFile(
                "s3://bucket/destination", 
                Void.class, 
                MyAvroType.class,
                new ParquetOutputFormat<MyAvroType>().getClass(), 
                job.getConfiguration());

Answer 2

暂时忽略S3；这是生产细节。您需要从更简单的问题“将我格式的本地文件转换为标准文件”开始。这是您可以通过针对小数据样本集的单元测试在本地实现的东西。

这在Spark中通常与Hadoop Mapduce相同：实现InputFormat的子类

有很多关于这方面的文档，以及堆栈溢出问题。在源树本身中有很多例子。