将Spark UDF用于将列值拆分为多个列

Question

将Spark UDF用于将列值拆分为多个列

scalaapache-sparkapache-spark-sqlapache-spark-2.0

3

我有一个数据框列名为“description”，其值格式如下：

ABC XXXXXXXXXXXX STORE NAME ABC TYPE1

我希望将它解析为以下三列：

|  mode |  type  |  store       |  description                           |
|------------------------------------------------------------------------|
|  ABC  |  TYPE1 |  STORE NAME  | ABC XXXXXXXXXXXX STORE NAME ABC TYPE1  |

我尝试了这里提供的方法。对于简单的UDF函数，它是有效的，但对于我编写的函数则不行。挑战在于 store 的值可能超过两个单词或没有固定数量的单词。

def myFunc1: (String => (String, String, String)) = { description =>
      var descripe = description.split(" ")
      val type = descripe(descripe.size - 1)
      descripe = description.substring(description.indexOf("ABC") + 4, description.lastIndexOf("ABC")).split(" ")
      val mode = descripe(0)
      descripe(0) = ""
      val store = descripe.mkString(" ").trim
      (mode, store, type)
    }

val schema = StructType(Array(
  StructField("mode", StringType, true),
  StructField("store", StringType, true),
  StructField("type", StringType, true)
))

val myUDF = udf(myFunc1, schema)

val test = pos.withColumn("test", myUDF(col("description")))
    test.printSchema()
val a =test.withColumn("mode", col("test").getItem("_1"))
    .withColumn("store", col("test").getItem("_2"))
    .withColumn("type", col("test").getItem("_3"))
    .drop(col("test"))

a.printSchema()
a.show(5, false)

当我执行时，出现以下错误：

18/10/06 21:38:02 ERROR Executor: 第5.0阶段任务0.0中的异常(TID 5) org.apache.spark.SparkException: 执行用户定义函数失败($anonfun$myFunc1$1$1: (string) => struct(mode:string,store:string,type:string)) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown Source) at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43) at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:395) at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:234) at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:228) at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:827) at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:827) at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323) at org.apache.spark.rdd.RDD.iterator(RDD.scala:287) at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87) at org.apache.spark.scheduler.Task.run(Task.scala:108) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:338) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) Caused by: java.lang.StringIndexOutOfBoundsException: 字符串索引超出范围：-4 at java.lang.String.substring(String.java:1967) at com.hasif.bank.track.trasaction.TransactionParser$$anonfun$myFunc1$1$1.apply(TransactionParser.scala:26) at com.hasif.bank.track.trasaction.TransactionParser$$anonfun$myFunc1$1$1.apply(TransactionParser.scala:22) ... 16 more 欢迎提供任何指导。

- Hasif Subair

1个回答

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- stack0114106 · Accepted Answer

检查一下这个。

scala> val df = Seq("ABC XXXXXXXXXXXX STORE NAME ABC TYPE1").toDF("desc")
df: org.apache.spark.sql.DataFrame = [desc: string]

scala> df.withColumn("mode",split('desc," ")(0)).withColumn("type",split('desc," ")(5)).withColumn("store",concat(split('desc," ")(2), lit(" "), split('desc," ")(3))).show(false)
+-------------------------------------+----+-----+----------+
|desc                                 |mode|type |store     |
+-------------------------------------+----+-----+----------+
|ABC XXXXXXXXXXXX STORE NAME ABC TYPE1|ABC |TYPE1|STORE NAME|
+-------------------------------------+----+-----+----------+


scala>

更新1：

scala> def splitStore(x:String):String=
     | return x.split(" ").drop(2).init.init.mkString(" ")
splitStore: (x: String)String

scala> val mysplitstore = udf(splitStore(_:String):String)
mysplitstore: org.apache.spark.sql.expressions.UserDefinedFunction = UserDefinedFunction(<function1>,StringType,Some(List(StringType)))

scala> val df2 = Seq("ABC XXXXXXXXXXXX STORE NAME XYZ ABC TYPE1").toDF("desc")
df2: org.apache.spark.sql.DataFrame = [desc: string]

scala> val df3 = df2.withColumn("length",split('desc," "))
df3: org.apache.spark.sql.DataFrame = [desc: string, length: array<string>]

scala> val df4 = df3.withColumn("mode",split('desc," ")(size('length)-2)).withColumn("type",split('desc," ")(size('length)-1)).withColumn("store",mysplitstore('desc))
df4: org.apache.spark.sql.DataFrame = [desc: string, length: array<string> ... 3 more fields]

scala> df4.drop('length).show(false)
+-----------------------------------------+----+-----+--------------+
|desc                                     |mode|type |store         |
+-----------------------------------------+----+-----+--------------+
|ABC XXXXXXXXXXXX STORE NAME XYZ ABC TYPE1|ABC |TYPE1|STORE NAME XYZ|
+-----------------------------------------+----+-----+--------------+


scala>