如何在 Spark/Scala 中使用另一列的分隔符来拆分列

Question

我还有一个与 split 函数相关的问题。 我是 Spark/Scala 新手。

下面是示例数据框 -

+-------------------+---------+
|             VALUES|Delimiter|
+-------------------+---------+
|       50000.0#0#0#|        #|
|          [email protected]@|        @|
|                 1$|        $|
|1000.00^Test_string|        ^|
+-------------------+---------+

我希望输出是-

+-------------------+---------+----------------------+
|VALUES             |Delimiter|split_values          |
+-------------------+---------+----------------------+
|50000.0#0#0#       |#        |[50000.0, 0, 0, ]     |
|[email protected]@          |@        |[0, 1000.0, ]         |
|1$                 |$        |[1, ]                 |
|1000.00^Test_string|^        |[1000.00, Test_string]|
+-------------------+---------+----------------------+

我尝试手动拆分它 -

dept.select(split(col("VALUES"),"#|@|\\$|\\^").show()

输出为-

+-----------------------+
|split(VALUES,#|@|\$|\^)|
+-----------------------+
|      [50000.0, 0, 0, ]|
|          [0, 1000.0, ]|
|                  [1, ]|
|   [1000.00, Test_st...|
+-----------------------+

但是我想为大数据集自动拉出分隔符。

Answer 1

您需要使用

expr

和

split()

来使分割动态化

df = spark.createDataFrame([("50000.0#0#0#","#"),("[email protected]@","@")],["VALUES","Delimiter"])
df = df.withColumn("split", F.expr("""split(VALUES, Delimiter)"""))
df.show()

+------------+---------+-----------------+
|      VALUES|Delimiter|            split|
+------------+---------+-----------------+
|50000.0#0#0#|        #|[50000.0, 0, 0, ]|
|   [email protected]@|        @|    [0, 1000.0, ]|
+------------+---------+-----------------+

Answer 2

编辑：请检查答案底部的 scala 版本。

您可以使用自定义的用户定义函数 (

pyspark.sql.functions.udf

) 来实现此目的。

from typing import List

from pyspark.sql import SparkSession
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType, ArrayType


def split_col(value: StringType, delimiter: StringType) -> List[str]:
    return str(value).split(str(delimiter))


udf_split = udf(lambda x, y: split_col(x, y), ArrayType(StringType()))

spark = SparkSession.builder.getOrCreate()

df = spark.createDataFrame([
    ('50000.0#0#0#', '#'), ('[email protected]@', '@'), ('1$', '$'), ('1000.00^Test_string', '^')
], schema='VALUES String, Delimiter String')

df = df.withColumn("split_values", udf_split(df['VALUES'], df['Delimiter']))

df.show(truncate=False)

输出

+-------------------+---------+----------------------+
|VALUES             |Delimiter|split_values          |
+-------------------+---------+----------------------+
|50000.0#0#0#       |#        |[50000.0, 0, 0, ]     |
|[email protected]@          |@        |[0, 1000.0, ]         |
|1$                 |$        |[1, ]                 |
|1000.00^Test_string|^        |[1000.00, Test_string]|
+-------------------+---------+----------------------+

请注意，

split_values

列包含字符串列表。您还可以更新

split_col

函数对值进行更多更改。

编辑：Scala 版本

import org.apache.spark.sql.functions.udf

import spark.implicits._

val data = Seq(("50000.0#0#0#", "#"), ("[email protected]@", "@"), ("1$", "$"), ("1000.00^Test_string", "^"))
var df = data.toDF("VALUES", "Delimiter")

val udf_split_col = udf {(x:String,y:String)=> x.split(y)}

df = df.withColumn("split_values", udf_split_col(df.col("VALUES"), df.col("Delimiter")))

df.show(false)

编辑2

为了避免正则表达式中使用特殊字符的问题，您可以在使用

split()

方法时使用 char 而不是 String，如下所示。

val udf_split_col = udf { (x: String, y: String) => x.split(y.charAt(0)) }

Answer 3

这是处理这个问题的另一种方法，使用sparksql

df.createOrReplaceTempView("test")

spark.sql("""select VALUES,delimiter,split(values,case when delimiter in ("$","^") then concat("\\",delimiter) else delimiter end) as split_value from test""").show(false)

请注意，我包含了 case when 语句来添加转义字符来处理“$”和“^”的情况，否则它不会拆分。

+-------------------+---------+----------------------+
|VALUES             |delimiter|split_value           |
+-------------------+---------+----------------------+
|50000.0#0#0#       |#        |[50000.0, 0, 0, ]     |
|[email protected]@          |@        |[0, 1000.0, ]         |
|1$                 |$        |[1, ]                 |
|1000.00^Test_string|^        |[1000.00, Test_string]|
+-------------------+---------+----------------------+

Answer 4

这是我最近的解决方案

import java.util.regex.Pattern
val split_udf = udf((value: String, delimiter: String) => value.split(Pattern.quote(delimiter), -1))
val solution = dept.withColumn("split_values", split_udf(col("VALUES"),col("Delimiter")))
solution.show(truncate = false)

它将跳过分隔符列中的特殊字符。 其他答案不适用于

("50000.0\\0\\0\\", "\\")

linusRian 的答案需要手动添加特殊字符

Answer 5

增强@dsk Answer以处理另外两个案例。

First Case:

要处理特殊字符

('$' ,'^')

，我们需要添加转义字符反斜杠

'\\'

作为分隔符列的前缀，如下所示..

import org.apache.spark.sql.functions.expr
val new_deptArray = dept.withColumn("split_values",expr("""split(VALUES, Concat('\\',Delimiter))"""))
new_deptArray.show(truncate = false)

这将是输出。

+-------------------+---------+----------------------+
|VALUES             |Delimiter|split_values          |
+-------------------+---------+----------------------+
|50000.0#0#0#       |#        |[50000.0, 0, 0, ]     |
|[email protected]@          |@        |[0, 1000.0, ]         |
|1$                 |$        |[1, ]                 |
|1000.00^Test_string|^        |[1000.00, Test_string]|
+-------------------+---------+----------------------+

Second Case:

请注意，很少有行在末尾有额外的空间，理想情况下我们不需要它们。所以这个新列

split_values

的类型是

Array<String>

，我们可以使用

array_remove

来删除这些额外的空格。

import org.apache.spark.sql.functions.{expr, array_remove}
val new_deptArray = dept.withColumn("split_values",array_remove(expr("""split(VALUES, Concat('\\',Delimiter))"""), ""))
new_deptArray.show(truncate = false)

它将给出这个输出。

+-------------------+---------+----------------------+
|VALUES             |Delimiter|split_values          |
+-------------------+---------+----------------------+
|50000.0#0#0#       |#        |[50000.0, 0, 0]       |
|[email protected]@          |@        |[0, 1000.0]           |
|1$                 |$        |[1]                   |
|1000.00^Test_string|^        |[1000.00, Test_string]|
+-------------------+---------+----------------------+

谢谢。