我有 CSV 檔案，如圖所示：
name,age,languages,experience
'Alice',31,['C ', 'Java'],2
'Bob',34,['Java', 'Python'],2
'Smith',35,['Ruby', 'Java'],3
'David',36,['C', 'Java', 'R']4
加載資料時，默認情況下所有列都作為字串加載。
scala> val df = spark.read.format("csv").option("header",true).load("data.csv")
df: org.apache.spark.sql.DataFrame = [name: string, age: string ... 2 more fields]
scala> df.show()
------- --- ------------------ ----------
| name|age| languages|experience|
------- --- ------------------ ----------
|'Alice'| 31| ['C ', 'Java']| 2|
| 'Bob'| 34|['Java', 'Python']| 2|
|'Smith'| 35| ['Ruby', 'Java']| 3|
|'David'| 36|['C', 'Java', 'R']| 4|
------- --- ------------------ ----------
scala> df.printSchema()
root
|-- name: string (nullable = true)
|-- age: string (nullable = true)
|-- languages: string (nullable = true)
|-- experience: string (nullable = true)
因此，我將自定義架構定義為String, Integer, Array,Integer資料型別：
import org.apache.spark.sql.types.{StructField, StructType, StringType, ArrayType, IntegerType}
val custom_schema = new StructType(Array(StructField("name", StringType), StructField("age", IntegerType), StructField("languages", ArrayType(StringType)), StructField("experience", IntegerType)))
當我使用自定義架構加載資料時，它拋出錯誤
運行命令后的終端截圖
scala> val df = spark.read.format("csv").option("header",true).schema(custom_schema).load("data.csv")
org.apache.spark.sql.AnalysisException: CSV data source does not support array<string> data type.
at org.apache.spark.sql.execution.datasources.DataSourceUtils$.$anonfun$verifySchema$1(DataSourceUtils.scala:67)
at org.apache.spark.sql.execution.datasources.DataSourceUtils$.$anonfun$verifySchema$1$adapted(DataSourceUtils.scala:65)
at scala.collection.Iterator.foreach(Iterator.scala:941)
at scala.collection.Iterator.foreach$(Iterator.scala:941)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1429)
at scala.collection.IterableLike.foreach(IterableLike.scala:74)
at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
at org.apache.spark.sql.types.StructType.foreach(StructType.scala:102)
at org.apache.spark.sql.execution.datasources.DataSourceUtils$.verifySchema(DataSourceUtils.scala:65)
at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:445)
at org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:326)
at org.apache.spark.sql.DataFrameReader.$anonfun$load$3(DataFrameReader.scala:308)
at scala.Option.getOrElse(Option.scala:189)
at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:308)
at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:240)
... 47 elided
如何通過將列作為陣列來加載資料以觸發資料幀？
uj5u.com熱心網友回復：
您可以在從檔案中讀取它后將其轉換為陣列[，]方法是使用洗掉方括號 ( , )regexp_replace并使用逗號 ( ,)拆分剩余的字串，split例如使用..
val df = spark.read.format("csv").option("header",true).load("data.csv")
val transformedDf = df.withColumn("languages",
split(
regexp_replace(col("languages"),"\\[|\\]",""),
","
)
)
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/365961.html
標籤：斯卡拉 文件 阿帕奇火花 apache-spark-sql 火花壳
上一篇：在PysPark中合并重疊區間