我在 HDFS 上有这个巨大的文件，它是我的数据库的一个提取物。例如:
1||||||1||||||||||||||0002||01||1999-06-01 16:18:38||||2999-12-31 00:00:00||||||||||||||||||||||||||||||||||||||||||||||||||||||||2||||0||W.ISHIHARA||||1999-06-01 16:18:38||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||19155||||||||||||||1||1||NBV||||||||||||||U||||||||N||||||||||||||||||||||
1||||||8||2000-08-25 00:00:00||||||||3||||0001||01||1999-06-01 16:26:16||||1999-06-01 17:57:10||||||||||300||||||PH||400||Yes||PH||0255097�`||400||||1||103520||||||1||4||10||||20||||||||||2||||0||S.OSARI||1961-10-05 00:00:00||1999-06-01 16:26:16||�o��������������||�o��������������||1||||����||||1||1994-01-24 00:00:00||2||||||75||1999-08-25 00:00:00||1999-08-25 00:00:00||0||1||||4||||||�l��������������||�o��������������||�l��������������||||�o��������������||NP||||�l��������������||�l��������������||||||5||19055||||||||||1||||8||1||NBV||||||||||||||U||||||||N||||||||||||||||||||||
- 文件大小:40GB
- 记录数:~120 000 000
- 字段数:112
- 字段分隔符:||
- 行分隔符:\n
- 编码:sjis
我想使用 pyspark(1.6 和 python 3)将这个文件加载到配置单元中。但我的工作一直失败。 这是我的代码:
toProcessFileDF = sc.binaryFiles("MyFile")\
.flatMap(lambda x: x[1].split(b'\n'))\
.map(lambda x: x.decode('sjis'))\
.filter(lambda x: x.count('|')==sepCnt*2)\
.map(lambda x: x.split('||'))\
.toDF(schema=tableSchema) #tableSchema is the schema retrieved from hive
toProcessFileDF.write.saveAsTable(tableName, mode='append')
我收到了几个错误，但其中包括 jave 143(内存错误)、心跳超时和内核已死。 (如果您需要确切的日志错误，请告诉我)。
这样做正确吗？也许有更聪明或更有效的方法。你能给我一些关于如何执行此操作的建议吗？
请您参考如下方法:
我发现 databrick csv 阅读器对此非常有用。
toProcessFileDF_raw = sqlContext.read.format('com.databricks.spark.csv')\
.options(header='false',
inferschema='false',
charset='shift-jis',
delimiter='\t')\
.load(toProcessFile)
不幸的是，我只能使用定界符选项来分割一个字符。因此，我的解决方案是用制表符拆分，因为我确定我的文件中没有任何内容。然后我可以在我的线上应用拆分。
这并不完美，但至少我有正确的编码并且我没有把所有东西都放在内存中。
