损坏的镶木地板文件答案

【问题标题】：Corrupted parquet file损坏的镶木地板文件
【发布时间】：2018-07-01 02:22:26
【问题描述】：

我在尝试使用 spark 读取 parquet 文件时遇到问题。 prquet文件是sqoop创建的：

sqoop import \
  --connect jdbc:teradata://<ip>/Database=<DB> \
  --connection-manager org.apache.sqoop.teradata.TeradataConnManager \
  --username <user> \
  --password <pass> \
  --table OFFERING \
  --target-dir /DWH/OFFERING \
  --as-parquetfile \
  --compress \
  --compression-codec org.apache.hadoop.io.compress.SnappyCodec \
  -m 8

hdfs 目录中的文件看起来是正确的：

[root@omm102 ~]# hdfs dfs -ls /DWH/OFFERING
Found 9 items
-rw-r--r--   3 hdfs hdfs          0 2018-01-19 20:44 /DWH/OFFERING/_SUCCESS
-rw-r--r--   3 hdfs hdfs       3630 2018-01-19 20:44 /DWH/OFFERING/part-m-00000
-rw-r--r--   3 hdfs hdfs       4046 2018-01-19 20:44 /DWH/OFFERING/part-m-00001
-rw-r--r--   3 hdfs hdfs       3146 2018-01-19 20:44 /DWH/OFFERING/part-m-00002
-rw-r--r--   3 hdfs hdfs       3703 2018-01-19 20:44 /DWH/OFFERING/part-m-00003
-rw-r--r--   3 hdfs hdfs       3065 2018-01-19 20:44 /DWH/OFFERING/part-m-00004
-rw-r--r--   3 hdfs hdfs       2972 2018-01-19 20:44 /DWH/OFFERING/part-m-00005
-rw-r--r--   3 hdfs hdfs       3405 2018-01-19 20:44 /DWH/OFFERING/part-m-00006
-rw-r--r--   3 hdfs hdfs       3091 2018-01-19 20:44 /DWH/OFFERING/part-m-00007

我也用 fstck 验证过：

[root@omm101 ~]# hdfs fsck /DWH/OFFERING -files
Connecting to namenode via http://omm101.xxx:<port>/fsck?ugi=root&files=1&path=%2FDWH%2FOFFERING
FSCK started by root (auth:SIMPLE) from /<ip> for path /DWH/OFFERING at Mon Jan 22 15:40:35 GST 2018
/DWH/OFFERING <dir>
/DWH/OFFERING/_SUCCESS 0 bytes, 0 block(s):  OK
/DWH/OFFERING/part-m-00000 3630 bytes, 1 block(s):  OK
/DWH/OFFERING/part-m-00001 4046 bytes, 1 block(s):  OK
/DWH/OFFERING/part-m-00002 3146 bytes, 1 block(s):  OK
/DWH/OFFERING/part-m-00003 3703 bytes, 1 block(s):  OK
/DWH/OFFERING/part-m-00004 3065 bytes, 1 block(s):  OK
/DWH/OFFERING/part-m-00005 2972 bytes, 1 block(s):  OK
/DWH/OFFERING/part-m-00006 3405 bytes, 1 block(s):  OK
/DWH/OFFERING/part-m-00007 3091 bytes, 1 block(s):  OK
Status: HEALTHY
 Total size:    27058 B
 Total dirs:    1
 Total files:   9
 Total symlinks:                0
 Total blocks (validated):      8 (avg. block size 3382 B)
 Minimally replicated blocks:   8 (100.0 %)
 Over-replicated blocks:        0 (0.0 %)
 Under-replicated blocks:       0 (0.0 %)
 Mis-replicated blocks:         0 (0.0 %)
 Default replication factor:    3
 Average block replication:     3.0
 Corrupt blocks:                0
 Missing replicas:              0 (0.0 %)
 Number of data-nodes:          5
 Number of racks:               1
FSCK ended at Mon Jan 22 15:40:35 GST 2018 in 2 milliseconds

但是当我启动 spark-shell (2.2.1) 并尝试阅读它时：

val offering = spark.read.parquet("/DWH/OFFERING/")

我收到以下错误：

java.io.IOException: Could not read footer for file: FileStatus{path=hdfs://mycluster/DWH/ACCOUNT_PARTY/part-m-00000; isDirectory=false; length=60604112; replication=0; blocksize=0; modification_time=0; access_time=0; owner=; group=; permission=rw-rw-rw-; isSymlink=false}
        at org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat$$anonfun$readParquetFootersInParallel$1.apply(ParquetFileFormat.scala:506)
        at org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat$$anonfun$readParquetFootersInParallel$1.apply(ParquetFileFormat.scala:493)
        at scala.collection.parallel.AugmentedIterableIterator$class.flatmap2combiner(RemainsIterator.scala:132)
        at scala.collection.parallel.immutable.ParVector$ParVectorIterator.flatmap2combiner(ParVector.scala:62)
        at scala.collection.parallel.ParIterableLike$FlatMap.leaf(ParIterableLike.scala:1072)
        at scala.collection.parallel.Task$$anonfun$tryLeaf$1.apply$mcV$sp(Tasks.scala:49)
        at scala.collection.parallel.Task$$anonfun$tryLeaf$1.apply(Tasks.scala:48)
        at scala.collection.parallel.Task$$anonfun$tryLeaf$1.apply(Tasks.scala:48)
        at scala.collection.parallel.Task$class.tryLeaf(Tasks.scala:51)
        at scala.collection.parallel.ParIterableLike$FlatMap.tryLeaf(ParIterableLike.scala:1068)
        at scala.collection.parallel.AdaptiveWorkStealingTasks$WrappedTask$class.compute(Tasks.scala:152)
        at scala.collection.parallel.AdaptiveWorkStealingForkJoinTasks$WrappedTask.compute(Tasks.scala:443)
        at scala.concurrent.forkjoin.RecursiveAction.exec(RecursiveAction.java:160)
        at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
        at scala.concurrent.forkjoin.ForkJoinTask.doJoin(ForkJoinTask.java:341)
        at scala.concurrent.forkjoin.ForkJoinTask.join(ForkJoinTask.java:673)
        at scala.collection.parallel.ForkJoinTasks$WrappedTask$class.sync(Tasks.scala:378)
        at scala.collection.parallel.AdaptiveWorkStealingForkJoinTasks$WrappedTask.sync(Tasks.scala:443)
        at scala.collection.parallel.ForkJoinTasks$class.executeAndWaitResult(Tasks.scala:426)
        at scala.collection.parallel.ForkJoinTaskSupport.executeAndWaitResult(TaskSupport.scala:56)
        at scala.collection.parallel.ParIterableLike$ResultMapping.leaf(ParIterableLike.scala:958)
        at scala.collection.parallel.Task$$anonfun$tryLeaf$1.apply$mcV$sp(Tasks.scala:49)
        at scala.collection.parallel.Task$$anonfun$tryLeaf$1.apply(Tasks.scala:48)
        at scala.collection.parallel.Task$$anonfun$tryLeaf$1.apply(Tasks.scala:48)
        at scala.collection.parallel.Task$class.tryLeaf(Tasks.scala:51)
        at scala.collection.parallel.ParIterableLike$ResultMapping.tryLeaf(ParIterableLike.scala:953)
        at scala.collection.parallel.AdaptiveWorkStealingTasks$WrappedTask$class.compute(Tasks.scala:152)
        at scala.collection.parallel.AdaptiveWorkStealingForkJoinTasks$WrappedTask.compute(Tasks.scala:443)
        at scala.concurrent.forkjoin.RecursiveAction.exec(RecursiveAction.java:160)
        at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
        at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
        at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
        at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
Caused by: java.lang.RuntimeException: hdfs://mycluster/DWH/ACCOUNT_PARTY/part-m-00000 is not a Parquet file. expected magic number at tail [80, 65, 82, 49] but found [48, 52, 53, 10]
        at org.apache.parquet.hadoop.ParquetFileReader.readFooter(ParquetFileReader.java:476)
        at org.apache.parquet.hadoop.ParquetFileReader.readFooter(ParquetFileReader.java:445)
        at org.apache.parquet.hadoop.ParquetFileReader.readFooter(ParquetFileReader.java:421)
        at org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat$$anonfun$readParquetFootersInParallel$1.apply(ParquetFileFormat.scala:499)
        ... 32 more

我不确定如何继续前进。

任何帮助将不胜感激。

EDIT1：

为了验证这是否与下面建议的压缩机制无关，我检查了下面的 sqoop：

sqoop import \
  --connect jdbc:teradata://<ip>/Database=<db> \
  --connection-manager org.apache.sqoop.teradata.TeradataConnManager \
  --username <usr> \
  --password <paswd> \
  --table OFFERING \
  --target-dir /DWH/TST/OFFERING \
  --as-parquetfile \
  -m 8

ant 然后又试着读了一遍：

val offering = spark.read.parquet("/DWH/TST/OFFERING/")

但结果是一样的

【问题讨论】：

您可能需要以不同的方式阅读活泼的压缩镶木地板
我认为这与阅读拼花无关。我已经启动了 spark-shell：./spark-shell --conf spark.io.compression.codec=org.apache.spark.io.SnappyCompressionCodec 并且，为了确保它被设置，还执行了 spark.conf.get("spark.io.compression.codec")。尝试执行时：val offering = spark.read.parquet("/DWH/OFFERING/") 我仍然遇到同样的错误。

标签： hadoop apache-spark hdfs sqoop parquet

【解决方案1】：

这里的准确答案很简单。以下错误是 100% 正确的：

Caused by: java.lang.RuntimeException: hdfs://mycluster/DWH/ACCOUNT_PARTY/part-m-00000 is not a Parquet file.

sqoop 生成的文件与 parquet 无关。这些只是 CSV 文件。即使我使用了--as-parquetfile 标志，这也没有任何意义，因为 Hortonworks Connector for Teradata 不支持 Parquet。有趣的是，sqoop 在导入过程中不会使用不适当的标志创建任何 WARN。

【讨论】：