私はScalaとSparkを初めて使用し、ローカルでcsvファイルを読み取ろうとしています（テスト用）。
val spark = org.apache.spark.sql.SparkSession.builder.master("local").appName("Spark CSV Reader").getOrCreate;
val topics_df = spark.read.format("csv").option("header", "true").load("path-to-file.csv")
topics_df.show(10)
ファイルは次のようになります。
+-----+--------------------+--------------------+
|topic| termindices| termweights|
+-----+--------------------+--------------------+
| 15|[21,31,51,108,101...|[0.0987100701,0.0...|
| 16|[42,25,121,132,55...|[0.0405490884,0.0...|
| 7|[1,23,38,7,63,0,1...|[0.1793091892,0.0...|
| 8|[13,40,35,104,153...|[0.0737646511,0.0...|
| 9|[2,10,93,9,158,18...|[0.1639456608,0.1...|
| 0|[28,39,71,46,123,...|[0.0867449145,0.0...|
| 1|[11,34,36,110,112...|[0.0729913664,0.0...|
| 17|[6,4,14,82,157,61...|[0.1583892199,0.1...|
| 18|[9,27,74,103,166,...|[0.0633899386,0.0...|
| 19|[15,81,289,218,34...|[0.1348582482,0.0...|
+-----+--------------------+--------------------+
と
ReadSchema: struct<topic:string,termindices:string,termweights:string>
termindices列は型であることになっているArray[Int]が、CSVに保存したときにそれはString（私は、データベースから引っ張っ場合、これは通常は問題ではありません）。
タイプを変換し、最終的にDataFrameを次のようにキャストするにはどうすればよいですか。
case class TopicDFRow(topic: Int, termIndices: Array[Int], termWeights: Array[Double])
変換を実行する準備ができている関数があります。
termIndices.substring(1, termIndices.length - 1).split(",").map(_.toInt)
私はudf他のいくつかの解決策を検討しましたが、上記の変換を実行するためのはるかにクリーンで高速な方法があるはずだと確信しています。どんな助けでも大歓迎です！
より効率的な組み込みのSpark関数を使用できる場合は、UDFを回避する必要があります。私の知る限り、提案された方法よりも良い方法はありません。文字列の最初と最後の文字を削除し、分割して変換します。
組み込み関数を使用すると、これは次のように実行できます。
df.withColumn("termindices", split($"termindices".substr(lit(2), length($"termindices")-2), ",").cast("array<int>"))
.withColumn("termweights", split($"termweights".substr(lit(2), length($"termweights")-2), ",").cast("array<double>"))
.as[TopicDFRow]
substr1インデックスベースの場合、最初の文字を削除するために2から開始します。2番目の引数は（終点ではなく）取る長さであるため、-2。
最後のコマンドは、データフレームをタイプのデータセットにキャストしTopicDFRowます。
この記事はインターネットから収集されたものであり、転載の際にはソースを示してください。
侵害の場合は、連絡してください[email protected]
コメントを追加