将数据集写入 Hive 时出现异常

【问题标题】：Getting Exception while writing Dataset to Hive将数据集写入 Hive 时出现异常
【发布时间】：2017-05-16 07:09:57
【问题描述】：

我正在尝试使用 Spark Java 将 DataSet 写入 Hive 数据库，但在此过程中我遇到了异常。

这是我的代码：

 Dataset<Row> data = spark.read().json(rdd).select("event.event_name");
 data.write().mode("overwrite").saveAsTable("telecom.t2");

这里，rdd 是流式 json 数据，我可以通过以下命令打印结果 data。

data.show();

但是当我尝试将此结果写入 Hive 数据库时，我没有收到任何异常，但是当我尝试打印这些值时，我在 Hive command line 中遇到异常。例如：

select * from telecom.t2;

例外是：

java.lang.reflect.InvocationTargetException
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.xerial.snappy.SnappyLoader.loadNativeLibrary(SnappyLoader.java:317)
    at org.xerial.snappy.SnappyLoader.load(SnappyLoader.java:219)
    at org.xerial.snappy.Snappy.<clinit>(Snappy.java:44)
    at parquet.hadoop.codec.SnappyDecompressor.decompress(SnappyDecompressor.java:62)
    at parquet.hadoop.codec.NonBlockedDecompressorStream.read(NonBlockedDecompressorStream.java:51)
    at java.io.DataInputStream.readFully(DataInputStream.java:195)
    at java.io.DataInputStream.readFully(DataInputStream.java:169)
    at parquet.bytes.BytesInput$StreamBytesInput.toByteArray(BytesInput.java:204)
    at parquet.column.values.dictionary.PlainValuesDictionary$PlainBinaryDictionary.<init>(PlainValuesDictionary.java:89)
    at parquet.column.values.dictionary.PlainValuesDictionary$PlainBinaryDictionary.<init>(PlainValuesDictionary.java:72)
    at parquet.column.Encoding$1.initDictionary(Encoding.java:89)
    at parquet.column.Encoding$4.initDictionary(Encoding.java:148)
    at parquet.column.impl.ColumnReaderImpl.<init>(ColumnReaderImpl.java:337)
    at parquet.column.impl.ColumnReadStoreImpl.newMemColumnReader(ColumnReadStoreImpl.java:66)
    at parquet.column.impl.ColumnReadStoreImpl.getColumnReader(ColumnReadStoreImpl.java:61)
    at parquet.io.RecordReaderImplementation.<init>(RecordReaderImplementation.java:270)
    at parquet.io.MessageColumnIO$1.visit(MessageColumnIO.java:134)
    at parquet.io.MessageColumnIO$1.visit(MessageColumnIO.java:99)
    at parquet.filter2.compat.FilterCompat$NoOpFilter.accept(FilterCompat.java:154)
    at parquet.io.MessageColumnIO.getRecordReader(MessageColumnIO.java:99)
    at parquet.hadoop.InternalParquetRecordReader.checkRead(InternalParquetRecordReader.java:137)
    at parquet.hadoop.InternalParquetRecordReader.nextKeyValue(InternalParquetRecordReader.java:208)
    at parquet.hadoop.ParquetRecordReader.nextKeyValue(ParquetRecordReader.java:201)
    at org.apache.hadoop.hive.ql.io.parquet.read.ParquetRecordReaderWrapper.<init>(ParquetRecordReaderWrapper.java:122)
    at org.apache.hadoop.hive.ql.io.parquet.read.ParquetRecordReaderWrapper.<init>(ParquetRecordReaderWrapper.java:85)
    at org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat.getRecordReader(MapredParquetInputFormat.java:72)
    at org.apache.hadoop.hive.ql.exec.FetchOperator$FetchInputFormatSplit.getRecordReader(FetchOperator.java:673)
    at org.apache.hadoop.hive.ql.exec.FetchOperator.getRecordReader(FetchOperator.java:323)
    at org.apache.hadoop.hive.ql.exec.FetchOperator.getNextRow(FetchOperator.java:445)
    at org.apache.hadoop.hive.ql.exec.FetchOperator.pushRow(FetchOperator.java:414)
    at org.apache.hadoop.hive.ql.exec.FetchTask.fetch(FetchTask.java:140)
    at org.apache.hadoop.hive.ql.Driver.getResults(Driver.java:1670)
    at org.apache.hadoop.hive.cli.CliDriver.processLocalCmd(CliDriver.java:233)
    at org.apache.hadoop.hive.cli.CliDriver.processCmd(CliDriver.java:165)
    at org.apache.hadoop.hive.cli.CliDriver.processLine(CliDriver.java:376)
    at org.apache.hadoop.hive.cli.CliDriver.executeDriver(CliDriver.java:736)
    at org.apache.hadoop.hive.cli.CliDriver.run(CliDriver.java:681)
    at org.apache.hadoop.hive.cli.CliDriver.main(CliDriver.java:621)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.apache.hadoop.util.RunJar.run(RunJar.java:221)
    at org.apache.hadoop.util.RunJar.main(RunJar.java:136)
Caused by: java.lang.UnsatisfiedLinkError: no snappyjava in java.library.path
    at java.lang.ClassLoader.loadLibrary(ClassLoader.java:1867)
    at java.lang.Runtime.loadLibrary0(Runtime.java:870)
    at java.lang.System.loadLibrary(System.java:1122)
    at org.xerial.snappy.SnappyNativeLoader.loadLibrary(SnappyNativeLoader.java:52)
    ... 48 more
Exception in thread "main" org.xerial.snappy.SnappyError: [FAILED_TO_LOAD_NATIVE_LIBRARY] null
    at org.xerial.snappy.SnappyLoader.load(SnappyLoader.java:229)
    at org.xerial.snappy.Snappy.<clinit>(Snappy.java:44)
    at parquet.hadoop.codec.SnappyDecompressor.decompress(SnappyDecompressor.java:62)
    at parquet.hadoop.codec.NonBlockedDecompressorStream.read(NonBlockedDecompressorStream.java:51)
    at java.io.DataInputStream.readFully(DataInputStream.java:195)
    at java.io.DataInputStream.readFully(DataInputStream.java:169)
    at parquet.bytes.BytesInput$StreamBytesInput.toByteArray(BytesInput.java:204)
    at parquet.column.values.dictionary.PlainValuesDictionary$PlainBinaryDictionary.<init>(PlainValuesDictionary.java:89)
    at parquet.column.values.dictionary.PlainValuesDictionary$PlainBinaryDictionary.<init>(PlainValuesDictionary.java:72)
    at parquet.column.Encoding$1.initDictionary(Encoding.java:89)
    at parquet.column.Encoding$4.initDictionary(Encoding.java:148)
    at parquet.column.impl.ColumnReaderImpl.<init>(ColumnReaderImpl.java:337)
    at parquet.column.impl.ColumnReadStoreImpl.newMemColumnReader(ColumnReadStoreImpl.java:66)
    at parquet.column.impl.ColumnReadStoreImpl.getColumnReader(ColumnReadStoreImpl.java:61)
    at parquet.io.RecordReaderImplementation.<init>(RecordReaderImplementation.java:270)
    at parquet.io.MessageColumnIO$1.visit(MessageColumnIO.java:134)
    at parquet.io.MessageColumnIO$1.visit(MessageColumnIO.java:99)
    at parquet.filter2.compat.FilterCompat$NoOpFilter.accept(FilterCompat.java:154)
    at parquet.io.MessageColumnIO.getRecordReader(MessageColumnIO.java:99)
    at parquet.hadoop.InternalParquetRecordReader.checkRead(InternalParquetRecordReader.java:137)
    at parquet.hadoop.InternalParquetRecordReader.nextKeyValue(InternalParquetRecordReader.java:208)
    at parquet.hadoop.ParquetRecordReader.nextKeyValue(ParquetRecordReader.java:201)
    at org.apache.hadoop.hive.ql.io.parquet.read.ParquetRecordReaderWrapper.<init>(ParquetRecordReaderWrapper.java:122)
    at org.apache.hadoop.hive.ql.io.parquet.read.ParquetRecordReaderWrapper.<init>(ParquetRecordReaderWrapper.java:85)
    at org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat.getRecordReader(MapredParquetInputFormat.java:72)
    at org.apache.hadoop.hive.ql.exec.FetchOperator$FetchInputFormatSplit.getRecordReader(FetchOperator.java:673)
    at org.apache.hadoop.hive.ql.exec.FetchOperator.getRecordReader(FetchOperator.java:323)
    at org.apache.hadoop.hive.ql.exec.FetchOperator.getNextRow(FetchOperator.java:445)
    at org.apache.hadoop.hive.ql.exec.FetchOperator.pushRow(FetchOperator.java:414)
    at org.apache.hadoop.hive.ql.exec.FetchTask.fetch(FetchTask.java:140)
    at org.apache.hadoop.hive.ql.Driver.getResults(Driver.java:1670)
    at org.apache.hadoop.hive.cli.CliDriver.processLocalCmd(CliDriver.java:233)
    at org.apache.hadoop.hive.cli.CliDriver.processCmd(CliDriver.java:165)
    at org.apache.hadoop.hive.cli.CliDriver.processLine(CliDriver.java:376)
    at org.apache.hadoop.hive.cli.CliDriver.executeDriver(CliDriver.java:736)
    at org.apache.hadoop.hive.cli.CliDriver.run(CliDriver.java:681)
    at org.apache.hadoop.hive.cli.CliDriver.main(CliDriver.java:621)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.apache.hadoop.util.RunJar.run(RunJar.java:221)
    at org.apache.hadoop.util.RunJar.main(RunJar.java:136)
2 Jan, 2017 12:02:40 PM WARNING: parquet.hadoop.ParquetRecordReader: Can not initialize counter due to context is not a instance of TaskInputOutputContext, but is org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl
2 Jan, 2017 12:02:40 PM INFO: parquet.hadoop.InternalParquetRecordReader: RecordReader initialized will read a total of 12 records.
2 Jan, 2017 12:02:40 PM INFO: parquet.hadoop.InternalParquetRecordReader: at row 0. reading next block
2 Jan, 2017 12:02:40 PM INFO: parquet.hadoop.InternalParquetRecordReader: block read in memory in 29 ms. row count = 12

【问题讨论】：

    标签： json hadoop apache-spark hive spark-streaming


    【解决方案1】：

    在调用saveAsTable 时，Spark 默认以parquet.snappy 格式保存数据，并且似乎您在 hive 库路径中没有 snappy。更改写入器格式（例如更改为 json）将不起作用，因为 Hive 需要使用此选项创建的表中的序列文件。

    但您可以在将数据保存为表格之前更改压缩算法：

    spark.conf.set("spark.sql.parquet.compression.codec", "gzip")

    默认情况下，Hive 上应该可以使用 Gzip 压缩，如果出现任何问题，您仍然可以在不压缩的情况下保存数据：

    spark.conf.set("spark.sql.parquet.compression.codec", "uncompressed")

    【讨论】：

    • 感谢@Mariusz 的回复，当我按照你所说的那样尝试时，我在 hive 中遇到了以下异常：* 失败并出现异常 java.io.IOException:java.io.IOException: hdfs://localhost :8020/user/hive/warehouse/telecom.db/t2/part-r-00000-0bd1c641-0e4d-4a8c-98a3-b8d31ecd4172.json 不是序列文件*
    • 创建表t2(event_name string) ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' WITH SERDEPROPERTIES ('path'='hdfs ://localhost:8020/user/hive/warehouse/telecom.db/t2'）存储为输入格式'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat'输出格式'org.apache.hadoop.hive。 ql.io.parquet.MapredParquetOutputFormat' LOCATION 'hdfs://localhost:8020/user/hive/warehouse/telecom.db/t2'
    • TBLPROPERTIES ('COLUMN_STATS_ACCURATE'='false', 'EXTERNAL'='FALSE', 'numFiles'='1', 'numRows'='-1', 'rawDataSize'='-1 ', 'spark.sql.sources.provider'='parquet', 'spark.sql.sources.schema.numParts'='1', 'spark.sql.sources.schema.part.0'='{\" type\":\"struct\",\"fields\":[{\"name\":\"event_name\",\"type\":\"string\",\"nullable\":true, \"metadata\":{}}]}', 'totalSize'='1134', 'transient_lastDdlTime'='1483342731') 耗时：0.578 秒，获取：23 行
    • 你说得对，Hive 在读取格式更改的表格时遇到了一些问题。尝试将压缩方法从 snappy 更改为 gzip - 请参阅我的更新答案。
    • 遇到同样的异常
    【解决方案2】：

    这个错误

    org.xerial.snappy.SnappyError: [FAILED_TO_LOAD_NATIVE_LIBRARY] null

    与 snappy 问题有关https://github.com/xerial/snappy-java/issues/6

    在此问题的最后评论中，有一个解决方法：

    unzip snappy-java-1.0.4.1.jar
cd org/xerial/snappy/native/Mac/x86_64/
cp libsnappyjava.jnilib libsnappyjava.dylib
cd ../../../../../..
cp snappy-java-1.0.4.1.jar snappy-java-1.0.4.1.jar.old
jar cf snappy-java-1.0.4.1.jar org

    【讨论】：

      猜你喜欢
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 2019-09-27
      • 1970-01-01
      • 1970-01-01
      • 2013-12-25
      相关资源
      最近更新 更多
      热门标签
      Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode