apache-spark - 在多核上运行时，Spark 作业中的 hadoop bz2 库失败

Question

我目前在使用 Spark 和读取 bz2 文件时遇到问题。我使用的是 Spark 1.2.0(为 hadoop 2.4 预构建，但文件目前在本地只读)。为了测试，有大约 1500 个文件，每个文件大约 50KB 大小。

下面的脚本 count_loglines.py 说明了这个问题:

 from pyspark import SparkConf, SparkContext
 spark_conf = SparkConf().setAppName("SparkTest")
 sc = SparkContext(conf=spark_conf)

 overall_log_lines = sc.textFile('/files/bzipped/*.log.bz2')
 line_count = overall_log_lines.count()
 print line_count

在一个核心上本地运行脚本，它按预期工作。

spark/bin/spark-submit --master local[1] count_log_lines.py

使用在 2 个内核上运行脚本

spark/bin/spark-submit --master local[2] count_log_lines.py

以 hadoop bzip2 库的错误消息结尾，例如

 : org.apache.spark.SparkException: Job aborted due to stage failure: Task 60 in stage 0.0 failed 1 times, most recent failure: Lost task 60.0 in stage 0.0 (TID 60, localhost): java.io.IOException: unexpected end of stream
    at org.apache.hadoop.io.compress.bzip2.CBZip2InputStream.bsGetBit(CBZip2InputStream.java:626)

当我事先解压缩文件，并读取未压缩的日志文件而不是 bzipped 文件时，即 sc.textFile('/files/unzipped/*.log') 脚本按预期工作，也在多核上运行。

我的问题:这里有什么问题？如果在多个核心上运行，为什么 Spark 作业不能正确读取 bz2 文件？

感谢您的帮助!

Answer 1

我不确定任一文本文件是否支持 bz2 文件。

您可能会查看 pyspark newAPIHadoopFile 或 hadoopfile API。如果拆分的 bz2 文件包含文本(例如日志)，则可以使用:

stdout = sc.newAPIHadoopFile(path="/HDFSpath/to/folder/containing/bz2/", inputFormatClass="org.apache.hadoop.mapreduce.lib.input.TextInputFormat", keyClass="org.apache.hadoop.io.Text", valueClass="org.apache.hadoop.io.Text", keyConverter=None, valueConverter=None, conf=None, batchSize=5)

来源: http://spark.apache.org/docs/1.2.0/api/python/pyspark.html

hadoopFile(path, inputFormatClass, keyClass, valueClass, keyConverter=None, valueConverter=None, conf=None, batchSize=0)

从 HDFS、本地文件系统(在所有节点上可用)或任何 Hadoop 支持的文件系统 URI 中读取具有任意键和值类的“旧”Hadoop InputFormat。该机制与 sc.sequenceFile 相同。

Hadoop 配置可以作为 Python 字典传入。这将转换为 Java 中的配置。

参数:
path – Hadoop 文件的路径
inputFormatClass – Hadoop InputFormat 的完全限定类名(例如“org.apache.hadoop.mapred.TextInputFormat”)
keyClass – 关键 Writable 类的完全限定类名(例如“org.apache.hadoop.io.Text”)
valueClass – 值 Writable 类的完全限定类名(例如“org.apache.hadoop.io.LongWritable”)
keyConverter –(默认无)
valueConverter –(默认无)
conf – Hadoop 配置，作为字典传入(默认无)
batchSize – 表示为单个 Java 对象的 Python 对象的数量。 (默认为 0，自动选择 batchSize)

或者

newAPIHadoopFile(path, inputFormatClass, keyClass, valueClass, keyConverter=None, valueConverter=None, conf=None, batchSize=0)

从 HDFS、本地文件系统(在所有节点上可用)或任何 Hadoop 支持的文件系统 URI 中读取具有任意键和值类的“新 API”Hadoop InputFormat。该机制与 sc.sequenceFile 相同。

Hadoop 配置可以作为 Python 字典传入。这将转换为 Java 中的配置

参数:
path – Hadoop 文件的路径
inputFormatClass – Hadoop InputFormat 的完全限定类名(例如“org.apache.hadoop.mapreduce.lib.input.TextInputFormat”)
keyClass – 关键 Writable 类的完全限定类名(例如“org.apache.hadoop.io.Text”)
valueClass – 值 Writable 类的完全限定类名(例如“org.apache.hadoop.io.LongWritable”)
keyConverter –(默认无)
valueConverter –(默认无)
conf – Hadoop 配置，作为字典传入(默认无)
batchSize – 表示为单个 Java 对象的 Python 对象的数量。 (默认为 0，自动选择 batchSize)

RGs，

钾