将具有多字符分隔符的单个字符串文本文件插入到 spark 数据框中

Question

Spark 新手，我边走边学。我有一个非常大的文本文件，其中的列由“|||||”分隔我想插入到火花数据框中。但是，该文件只是一个行字符串。该文件如下所示：

col1|||||col2|||||col3|||||col4|||||col5|||||col1|||||col2|||||col3...

所以第 1 列到第 5 列基本上只是在一行中循环。我试图在每 5 个 "|||||" 之后插入一个新行使用 sed 命令通过：

 sed -r 's/([^|||||]*|||||){5}/&\n/g'

其中大部分工作但最终由于某种原因无法正常工作。我怀疑 col4（这是一个巨大的文本字段）在这方面造成了一些问题，但我不知道它为什么这样做。

现在，当我通过以下方式将单行文本文件读入 spark 时：

val df = spark.read.textFile(file) 

这会将所有内容放在一列中，我想将其拆分为 5 列，并让数据框在每 5 列之后“包装”字符串。

我的目标是让它变成这样：

+--------------------+---------------+--------------------+--------------------+--------------------+
|                col1|           col2|                col3|                col4|                col5|
+--------------------+---------------+--------------------+--------------------+--------------------+
|                 val|            val|                 val|                 val|                 val|
|                 val|            val|                 val|                 val|                 val|
+--------------------+---------------+--------------------+--------------------+--------------------+

所以我的问题是：因为我的文件只是一个大字符串，有没有办法让数据框在 5 列后输入新的记录/行？

Answer 1

这是您第一个问题的解决方案。 通常你会读为一个普通的文本文件，然后使用 split 方法将行转换为列。

df.withColumn("tmp", split($"value", "|||||")).select(
  $"tmp".getItem(0).as("first"),
  $"tmp".getItem(1).as("second"),
  $"tmp".getItem(2).as("third")
).drop("tmp")

关于你的第二个问题。您可以使用此正则表达式来匹配模式：

(([a-z0-9A-Z]+)(\|\|\|\|\|)([a-z0-9A-Z]+)(\|\|\|\|\ |)([a-z0-9A-Z]+)(\|\|\|\|\|)([a-z0-9A-Z]+)(\|\|\|\|\|) )

如果您有足够的内存，您可以读取所有文件，然后使用此模式提取其中的部分。

如果不是，那么你必须逐字节读取它，看看你是否匹配这个模式。

祝你好运！

Answer 2

如果文件很大，只有一行，那么请使用 Perl 解决方案。 Perl 变量可以存储文件内容（甚至以 GB 为单位），您可以轻松管理。您在 perl 本身中完成所有预处理。看看以下是否适合你

> cat 5cols
col1|||||col2|||||col3|||||col4|||||col5|||||col1|||||col2|||||col3|||||col4|||||col5|||||col1|||||col2|||||col3|||||col4|||||col5|||||col1|||||col2|||||col3|||||col4|||||col5|||||col1|||||col2|||||col3|||||col4|||||col5|||||col1|||||col2|||||col3|||||col4|||||col5|||||col1|||||col2|||||col3|||||col4|||||col5|||||col1|||||col2|||||col3|||||col4|||||col5|||||col1|||||col2|||||col3|||||col4|||||col5|||||col1|||||col2|||||col3|||||col4|||||col5|||||

> perl -e ' BEGIN {$x=qx(cat 5cols);while($x=~m/([^|]+?)(?=[|]{5})/g){ print "$1,\n"} exit } ' | xargs -n5 | sed 's/,$//g'
col1, col2, col3, col4, col5
col1, col2, col3, col4, col5
col1, col2, col3, col4, col5
col1, col2, col3, col4, col5
col1, col2, col3, col4, col5
col1, col2, col3, col4, col5
col1, col2, col3, col4, col5
col1, col2, col3, col4, col5
col1, col2, col3, col4, col5
col1, col2, col3, col4, col5

>

将上述输出重定向到另一个 csv 文件。现在，您可以将 spark.csv 读取为具有 5 列的常规 csv 文件