apache-spark - 读取spark中的字节列

Question

我有一个数据集，其中包含采用未知(且不友好)编码的 ID 字段。我可以使用普通 python 读取单列，并验证多个数据集中的值是否不同且一致(即它可以用作连接的主键)。

使用 spark.read.csv 加载文件时，spark 似乎正在将该列转换为 utf-8。但是，某些多字节序列会转换为 Unicode 字符 U+FFFD REPLACMENT CHARACTER。(十六进制的 EF BF BD)。

有没有办法强制 Spark 将列读取为字节而不是字符串？

下面是一些可用于重新创建我的问题的代码(让 a 列作为 ID 字段):

使用示例数据创建文件

data = [
    (bytes(b'\xba\xed\x85\x8e\x91\xd4\xc7\xb0'), '1', 'a'),
    (bytes(b'\xba\xed\x85\x8e\x91\xd4\xc7\xb1'), '2', 'b'),
    (bytes(b'\xba\xed\x85\x8e\x91\xd4\xc7\xb2'), '3', 'c')
]

with open('sample.csv', 'wb') as f:
    header = ["a", "b", "c"]
    f.write(",".join(header)+"\n")
    for d in data:
        f.write(",".join(d) + "\n")

使用 Pandas 阅读

import pandas as pd
df = pd.read_csv("sample.csv", converters={"a": lambda x: x.encode('hex')})
print(df)
#                  a  b  c
#0  baed858e91d4c7b0  1  a
#1  baed858e91d4c7b1  2  b
#2  baed858e91d4c7b2  3  c

尝试使用 Spark 读取同一文件

spark_df = spark.read.csv("sample.csv", header=True)
spark_df.show()
#+-----+---+---+
#|a    |b  |c  |
#+-----+---+---+
#|�텎��ǰ|1  |a  |
#|�텎��DZ|2  |b  |
#|�텎��Dz|3  |c  |
#+-----+---+---+

哎呀!好的，那么转换为 hex 怎么样？

import pyspark.sql.functions as f
spark_df.withColumn("a", f.hex("a")).show(truncate=False)
#+----------------------------+---+---+
#|a                           |b  |c  |
#+----------------------------+---+---+
#|EFBFBDED858EEFBFBDEFBFBDC7B0|1  |a  |
#|EFBFBDED858EEFBFBDEFBFBDC7B1|2  |b  |
#|EFBFBDED858EEFBFBDEFBFBDC7B2|3  |c  |
#+----------------------------+---+---+

(在此示例中，值是不同的，但在我的较大文件中并非如此)

如您所见，这些值接近，但某些字节已被EFBFBD替换

有什么方法可以读取 Spark 中的文件(也许使用 rdd？)，以便我的输出看起来像 pandas 版本:

#+----------------+---+---+
#|a               |b  |c  |
#+----------------+---+---+
#|baed858e91d4c7b0|1  |a  |
#|baed858e91d4c7b1|2  |b  |
#|baed858e91d4c7b2|3  |c  |
#+----------------+---+---+

我尝试转换为 byte 并指定架构，以便该列为 ByteType()，但这不起作用。

编辑

我使用的是 Spark v 2.1。

Answer 1

问题的根源在于分隔文件不太适合二进制数据。

如果文本有已知的一致编码，请使用charset 选项。请参阅https://github.com/databricks/spark-csv#features (我不知道 2.x 文档中描述分隔阅读选项的好地方，所以我仍然回到 1.x 文档)。我建议尝试使用 8 位 ASCII，例如 ISO-8859-1 或 US-ASCII。

如果没有这样的编码，您需要将输入转换为不同的格式，例如，对第一列进行 base64 编码，或者操作读取的数据以将其恢复为您需要的格式。