我正在尝试使用 spark 1.6 读取 CSV 文件
s.no|Name$id|designation|salry
1 |abc$12 |xxx |yyy
val df = spark.read.format("csv")
.option("header","true")
.option("delimiter","|")
.load("path")
如果我用 $ 添加分隔符，则会抛出错误，允许使用一个分隔符
【问题讨论】：
标签： apache-spark apache-spark-sql
您可以在使用主分隔符从源读取数据帧后应用操作（我将“|”称为主分隔符以便更好地理解）。
您可以执行以下操作：
sc 是 Sparksession
val inputDF = sc.read.option("inferSchema", "true")
.option("header", "true")
.option("delimiter", "|")
.csv("/path/to/your/file")
val modifiedDF = inputDF
.withColumn("Name", split(inputDF.col("Name$id"), "\\$")(0))
.withColumn("id", split(inputDF.col("Name$id"), "\\$")(1)).drop("Name$id")
modifiedDF.show(false) will give you the required output
虽然这可能会导致数据被错误地拆分，以防数据中有有效的“$”符号被误认为是分隔符。因此，您应该在这些情况下采取预防措施。
有一个库，不记得它的名称，但它可能是唯一性的，它使您可以选择将多个符号视为单个分隔符，例如 #@ 作为分隔符。如果您的用例是每列的多个分隔符，您可以搜索一下。
【讨论】：
请问您为什么使用 spark 1.6？ 无论如何，读取 csv 格式时只允许使用一个分隔符。
如果它是您知道的特定列，其中有一列的值格式为：name$id
也许尝试在该列上运行一些逻辑并获得一个包含 2 个新列的 df
设置 df
al df = sc.parallelize(a).toDF("nameid")
df: org.apache.spark.sql.DataFrame = [nameid: string]
试试这样的：
df.withColumn("name",substring_index(col("nameid"), "$", 1)).withColumn("id", substring_index(col("nameid"), "$", -1)).show
和输出
+-------+----+---+
| nameid|name| id|
+-------+----+---+
|name$id|name| id|
+-------+----+---+
之后您也可以删除原始列
希望这有帮助
【讨论】：
回答了一个不同的问题，但在这里重复。此处发布的其他解决方案假设那些特定的分隔符出现在特定的位置。但是，我假设最坏的情况是这些分隔符可以随机出现在任何地方。
使用 pyspark 举例说明下面的假设情况，其中分隔符是 |和 -
注意：我们正在使用 split ，这意味着它将拆分所有内容，例如2000-12-31 是一个值，但它将被拆分。因此，我们应该非常确定在数据中永远不会发生这种情况。作为一般建议，永远不要接受这些类型的文件，因为会有意外发生。
示例数据的外观：- 在本例中，我们的目录中有 2 个文件，其中 |和 - 作为分隔符随机出现
# Create RDD. Basically read as simple text file.
# sc is spark context
rddRead = sc.textFile("/mnt/adls/RI_Validation/ReadMulktipleDelimerFile/Sample1/")
rddRead.collect() # For debugging
import re # Import for usual python regex
# Create another rdd using simple string opertaions. This will be similar to list of lists.
# Give regex expression to split your string based on anticipated delimiters (this could be dangerous
# if those delimiter occur as part of value. e.g.: 2021-12-31 is a single value in reality.
# But this a price we have to pay for not having good data).
# For each iteration, k represents 1 element which would eventually become 1 row (e.g. A|33-Mech)
rddSplit = rddRead.map(lambda k: re.split("[|-]+", k)) # Anticipated delimiters are | OR - in this case.
rddSplit.collect() # For debugging
# This block is applicable only if you have headers
lsHeader = rddSplit.first() # Get First element from rdd as header.
print(lsHeader) # For debugging
print()
# Remove rows representing header. (Note: Have assumed name of all columns in
# all files are same. If not, then will have to filter by manually specifying
#all of them which would be a nightmare from pov of good code as well as maintenance)
rddData = rddSplit.filter(lambda x: x != lsHeader)
rddData.collect() # For debugging
# Convert rdd to spark dataframe
# Utilise the header we got in earlier step. Else can give our own headers.
dfSpark = rddData.toDF(lsHeader)
dfSpark.display() # For debugging
【讨论】：