java：从一个文件中读取json，转换成orc并写入一个文件

7fhtutme 于 2021-06-26 发布在 Hive

关注(0)|答案(1)|浏览(841)

我需要自动化json到orc的转换过程。通过使用apache的orc工具包，我几乎可以做到这一点，除了jsonreader is不处理Map类型并抛出异常。所以，下面的代码可以工作，但不能处理map类型。

Path hadoopInputPath = new Path(input);
    try (RecordReader recordReader = new JsonReader(hadoopInputPath, schema, hadoopConf)) { // throws when schema contains Map type
        try (Writer writer = OrcFile.createWriter(new Path(output), OrcFile.writerOptions(hadoopConf).setSchema(schema))) {
            VectorizedRowBatch batch = schema.createRowBatch();
            while (recordReader.nextBatch(batch)) {
                writer.addRowBatch(batch);
            }
        }
    }

因此，我开始考虑使用hive类进行json到orc的转换，这有一个额外的优势，将来我可以转换成其他格式，比如avro，只需稍作代码更改。但是，我不确定使用hive类实现这一点的最佳方法是什么。具体来说，不清楚如何将hcatrecord写入如下所示的文件。

HCatRecordSerDe hCatRecordSerDe = new HCatRecordSerDe();
    SerDeUtils.initializeSerDe(hCatRecordSerDe, conf, tblProps, null);

    OrcSerde orcSerde = new OrcSerde();
    SerDeUtils.initializeSerDe(orcSerde, conf, tblProps, null);

    Writable orcOut = orcSerde.serialize(hCatRecord, hCatRecordSerDe.getObjectInspector());
    assertNotNull(orcOut);

    InputStream input = getClass().getClassLoader().getResourceAsStream("test.json.snappy");
    SnappyCodec compressionCodec = new SnappyCodec();
    try (CompressionInputStream inputStream = compressionCodec.createInputStream(input)) {
        LineReader lineReader = new LineReader(new InputStreamReader(inputStream, Charsets.UTF_8));
        String jsonLine = null;
        while ((jsonLine = lineReader.readLine()) != null) {
            Writable jsonWritable = new Text(jsonLine);
            DefaultHCatRecord hCatRecord = (DefaultHCatRecord) jsonSerDe.deserialize(jsonWritable);
            // TODO: Write ORC to file????
        }
    }

任何关于如何完成以上代码的想法，或者用更简单的方法将json转换为orc的想法都将受到极大的赞赏。

Java Hive JSON apache orc

来源：https://stackoverflow.com/questions/46290348/java-read-json-from-a-file-convert-to-orc-and-write-to-a-file

1条答案

按热度按时间

cyej8jka1#

根据cricket\u 007的建议，我最终使用了spark库：
maven依赖项（有一些排除项可以让maven duplicate finder插件满意）：

<properties>
        <dep.jackson.version>2.7.9</dep.jackson.version>
        <spark.version>2.2.0</spark.version>
        <scala.binary.version>2.11</scala.binary.version>
    </properties>

    <dependency>
        <groupId>com.fasterxml.jackson.module</groupId>
        <artifactId>jackson-module-scala_${scala.binary.version}</artifactId>
        <version>${dep.jackson.version}</version>
        <exclusions>
            <exclusion>
                <groupId>com.google.guava</groupId>
                <artifactId>guava</artifactId>
            </exclusion>
        </exclusions>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-hive_${scala.binary.version}</artifactId>
        <version>${spark.version}</version>
        <exclusions>
            <exclusion>
                <groupId>log4j</groupId>
                <artifactId>apache-log4j-extras</artifactId>
            </exclusion>
            <exclusion>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-client</artifactId>
            </exclusion>
            <exclusion>
                <groupId>net.java.dev.jets3t</groupId>
                <artifactId>jets3t</artifactId>
            </exclusion>
            <exclusion>
                <groupId>com.google.code.findbugs</groupId>
                <artifactId>jsr305</artifactId>
            </exclusion>
            <exclusion>
                <groupId>stax</groupId>
                <artifactId>stax-api</artifactId>
            </exclusion>
            <exclusion>
                <groupId>org.objenesis</groupId>
                <artifactId>objenesis</artifactId>
            </exclusion>
        </exclusions>
    </dependency>

java代码概要：

SparkConf sparkConf = new SparkConf()
    .setAppName("Converter Service")
    .setMaster("local[*]");

SparkSession sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate();

// read input data
Dataset<Row> events = sparkSession.read()
    .format("json")
    .schema(inputConfig.getSchema()) // StructType describing input schema
    .load(inputFile.getPath());

// write data out
DataFrameWriter<Row> frameWriter = events
    .selectExpr(
        // useful if you want to change the schema before writing it to ORC, e.g. ["`col1` as `FirstName`", "`col2` as `LastName`"]
        JavaConversions.asScalaBuffer(outputSchema.getColumns()))
    .write()
    .options(ImmutableMap.of("compression", "zlib"))
    .format("orc")
    .save(outputUri.getPath());

希望这能帮助别人开始。

赞(0）回复(0）举报 2021-06-26

我来回答

java：从一个文件中读取json，转换成orc并写入一个文件

1条答案

相关问题

热门标签

最新问答