在云数据融合中读取和转换 parquet 文件

Question

尝试在云数据融合中摄取和转换 parquet 文件。我可以看到我可以使用 GCS 插件摄取镶木地板文件。但是当我想使用 wrangler 插件对其进行转换时，我看不到任何这样做的能力。牧马人插件是否有这种能力，还是我应该考虑另一种方法？顺便说一句，我刚刚部署了我的管道以查看我是否能够从 GCS 摄取镶木地板文件，但我在日志中看到了这个错误：

java.lang.NoClassDefFoundError: org/xerial/snappy/Snappy
    at org.apache.parquet.hadoop.codec.SnappyDecompressor.decompress(SnappyDecompressor.java:62) ~[parquet-hadoop-1.8.3.jar:1.8.3]
    at org.apache.parquet.hadoop.codec.NonBlockedDecompressorStream.read(NonBlockedDecompressorStream.java:51) ~[parquet-hadoop-1.8.3.jar:1.8.3]
    at java.io.DataInputStream.readFully(DataInputStream.java:195) ~[na:1.8.0_275]
    at java.io.DataInputStream.readFully(DataInputStream.java:169) ~[na:1.8.0_275]
    at org.apache.parquet.bytes.BytesInput$StreamBytesInput.toByteArray(BytesInput.java:204) ~[parquet-encoding-1.8.3.jar:1.8.3]
    at org.apache.parquet.column.values.dictionary.PlainValuesDictionary$PlainBinaryDictionary.<init>(PlainValuesDictionary.java:89) ~[parquet-column-1.8.3.jar:1.8.3]
    at org.apache.parquet.column.values.dictionary.PlainValuesDictionary$PlainBinaryDictionary.<init>(PlainValuesDictionary.java:72) ~[parquet-column-1.8.3.jar:1.8.3]
    at org.apache.parquet.column.Encoding$1.initDictionary(Encoding.java:90) ~[parquet-column-1.8.3.jar:1.8.3]
    at org.apache.parquet.column.Encoding$4.initDictionary(Encoding.java:149) ~[parquet-column-1.8.3.jar:1.8.3]
    at org.apache.parquet.column.impl.ColumnReaderImpl.<init>(ColumnReaderImpl.java:343) ~[parquet-column-1.8.3.jar:1.8.3]
    at org.apache.parquet.column.impl.ColumnReadStoreImpl.newMemColumnReader(ColumnReadStoreImpl.java:82) ~[parquet-column-1.8.3.jar:1.8.3]
    at org.apache.parquet.column.impl.ColumnReadStoreImpl.getColumnReader(ColumnReadStoreImpl.java:77) ~[parquet-column-1.8.3.jar:1.8.3]
    at org.apache.parquet.io.RecordReaderImplementation.<init>(RecordReaderImplementation.java:270) ~[parquet-column-1.8.3.jar:1.8.3]
    at org.apache.parquet.io.MessageColumnIO$1.visit(MessageColumnIO.java:135) ~[parquet-column-1.8.3.jar:1.8.3]
    at org.apache.parquet.io.MessageColumnIO$1.visit(MessageColumnIO.java:101) ~[parquet-column-1.8.3.jar:1.8.3]
    at org.apache.parquet.filter2.compat.FilterCompat$NoOpFilter.accept(FilterCompat.java:154) ~[parquet-column-1.8.3.jar:1.8.3]
    at org.apache.parquet.io.MessageColumnIO.getRecordReader(MessageColumnIO.java:101) ~[parquet-column-1.8.3.jar:1.8.3]
    at org.apache.parquet.hadoop.InternalParquetRecordReader.checkRead(InternalParquetRecordReader.java:140) ~[parquet-hadoop-1.8.3.jar:1.8.3]
    at org.apache.parquet.hadoop.InternalParquetRecordReader.nextKeyValue(InternalParquetRecordReader.java:214) ~[parquet-hadoop-1.8.3.jar:1.8.3]
    at org.apache.parquet.hadoop.ParquetRecordReader.nextKeyValue(ParquetRecordReader.java:227) ~[parquet-hadoop-1.8.3.jar:1.8.3]
    at io.cdap.plugin.format.parquet.input.PathTrackingParquetInputFormat$ParquetRecordReader.nextKeyValue(PathTrackingParquetInputFormat.java:76) ~[1614054281928-0/:na]
    at io.cdap.plugin.format.input.PathTrackingInputFormat$TrackingRecordReader.nextKeyValue(PathTrackingInputFormat.java:136) ~[na:na]
    at org.apache.hadoop.mapreduce.lib.input.CombineFileRecordReaderWrapper.nextKeyValue(CombineFileRecordReaderWrapper.java:90) ~[hadoop-mapreduce-client-core-2.9.2.jar:na]
    at org.apache.hadoop.mapreduce.lib.input.CombineFileRecordReader.nextKeyValue(CombineFileRecordReader.java:65) ~[hadoop-mapreduce-client-core-2.9.2.jar:na]
    at org.apache.spark.rdd.NewHadoopRDD$$anon$1.hasNext(NewHadoopRDD.scala:214) ~[spark-core_2.11-2.3.4.jar:2.3.4]
    at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37) ~[spark-core_2.11-2.3.4.jar:2.3.4]
    at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:439) ~[scala-library-2.11.8.jar:na]
    at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:439) ~[scala-library-2.11.8.jar:na]
    at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:439) ~[scala-library-2.11.8.jar:na]
    at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:439) ~[scala-library-2.11.8.jar:na]
    at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:439) ~[scala-library-2.11.8.jar:na]
    at org.apache.spark.internal.io.SparkHadoopWriter$$anonfun$4.apply(SparkHadoopWriter.scala:130) ~[spark-core_2.11-2.3.4.jar:2.3.4]
    at org.apache.spark.internal.io.SparkHadoopWriter$$anonfun$4.apply(SparkHadoopWriter.scala:129) ~[spark-core_2.11-2.3.4.jar:2.3.4]
    at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1415) ~[spark-core_2.11-2.3.4.jar:2.3.4]
    at org.apache.spark.internal.io.SparkHadoopWriter$.org$apache$spark$internal$io$SparkHadoopWriter$$executeTask(SparkHadoopWriter.scala:141) [spark-core_2.11-2.3.4.jar:2.3.4]
    at org.apache.spark.internal.io.SparkHadoopWriter$$anonfun$3.apply(SparkHadoopWriter.scala:83) [spark-core_2.11-2.3.4.jar:2.3.4]
    at org.apache.spark.internal.io.SparkHadoopWriter$$anonfun$3.apply(SparkHadoopWriter.scala:78) [spark-core_2.11-2.3.4.jar:2.3.4]
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87) [spark-core_2.11-2.3.4.jar:2.3.4]
    at org.apache.spark.scheduler.Task.run(Task.scala:109) [spark-core_2.11-2.3.4.jar:2.3.4]
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345) [spark-core_2.11-2.3.4.jar:2.3.4]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_275]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_275]
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_275]
Caused by: java.lang.ClassNotFoundException: org.xerial.snappy.Snappy
    at java.net.URLClassLoader.findClass(URLClassLoader.java:382) ~[na:1.8.0_275]
    at io.cdap.cdap.common.lang.InterceptableClassLoader.findClass(InterceptableClassLoader.java:44) ~[na:na]
    at java.lang.ClassLoader.loadClass(ClassLoader.java:418) ~[na:1.8.0_275]
    at java.lang.ClassLoader.loadClass(ClassLoader.java:351) ~[na:1.8.0_275]
    ... 43 common frames omitted

我是否需要在我的集群上安装一个特定的模块（这可能吗）？

Answer 1

您使用的是哪个版本的 Dataproc？

Dataproc 2.0 的某些版本似乎没有包含 Hadoop 下的 Snappy 库。

作为一种解决方法，您可以将 snappy jar 复制到以下目录：

sudo cp /usr/lib/hive/lib/snappy-java-*.jar /usr/lib/hadoop-mapreduce/
sudo cp /usr/lib/hive/lib/snappy-java-*.jar /usr/lib/hadoop/lib
sudo cp /usr/lib/hive/lib/snappy-java-*.jar /usr/lib/hadoop-yarn/lib

这可以通过 Data Fusion Compute Profile 在 Dataproc init-action [1] 脚本中完成，如下所示：

在 Data Fusion 中，转到：系统管理 -> 配置 -> 系统计算配置文件 ->（新建）-> 高级设置 -> 初始化操作。 否则，在已部署的管道上，在工作室中，转到配置 -> 计算配置 ->（选择配置文件）-> 自定义 -> 高级设置 -> 初始化操作。

注意：这些选项在 Developer 实例中不可用，仅在 Basic/Enterprise 中可用。

[1]https://cloud.google.com/dataproc/docs/concepts/configuring-clusters/init-actions