スパークは以下

Question

は私のcsvファイルの内容の最初のローにnull値を持つ列を読み取れません：

A1,B1,C1 
A2,B2,C2,D1 
A3,B3,C3,D2,E1 
A4,B4,C4,D3 
A5,B5,C5,,E2 

ので、5列が、最初の行の唯一の3つの値があります。

私は、次のコマンドを使用して、それを読む：

val csvDF : DataFrame = spark.read 
.option("header", "false") 
.option("delimiter", ",") 
.option("inferSchema", "false") 
.csv("file.csv") 

そして、次は私がcsvDF.showを（使用して得るものです）

+---+---+---+ 
|_c0|_c1|_c2| 
+---+---+---+ 
| A1| B1| C1| 
| A2| B2| C2| 
| A3| B3| C3| 
| A4| B4| C4| 
| A5| B5| C5| 
+---+---+---+ 

を私はすべての列内のすべてのデータを読み取ることができますどのように？

Answer 1

あなたは（別の区切り文字を使用することにより、例えば）1列のみを持つデータセットとしてそれを読むことができます。

var df = spark.read.format("csv").option("delimiter",";").load("test.csv") 
df.show() 

+--------------+ 
|   _c0| 
+--------------+ 
|  A1,B1,C1| 
| A2,B2,C2,D1| 
|A3,B3,C3,D2,E1| 
| A4,B4,C4,D3| 
| A5,B5,C5,,E2| 
+--------------+ 

その後、あなたはこれがヌルを追加し、手動で5であなたのコラムを分割するthis answerを使用することができます要素が存在しない値：

var csvDF = df.withColumn("_tmp",split($"_c0",",")).select(
    $"_tmp".getItem(0).as("col1"), 
    $"_tmp".getItem(1).as("col2"), 
    $"_tmp".getItem(2).as("col3"), 
    $"_tmp".getItem(3).as("col4"), 
    $"_tmp".getItem(4).as("col5") 
) 
csvDF.show() 

+----+----+----+----+----+ 
|col1|col2|col3|col4|col5| 
+----+----+----+----+----+ 
| A1| B1| C1|null|null| 
| A2| B2| C2| D1|null| 
| A3| B3| C3| D2| E1| 
| A4| B4| C4| D3|null| 
| A5| B5| C5| | E2| 
+----+----+----+----+----+ 

Answer 2

基本的にはあなたのCSVファイルが正しく意味でフォーマットされていません、あなたがしたい場合は、それが必要とされ、各行の列の数と同じ数を持っていないことspark.read.csvとお読みください。ただし、代わりにspark.read.textFileとそれを読み取って、各行を解析することができます。

私が理解しているように、事前に列数を知っていないので、コードで任意の数の列を処理する必要があります。これを行うには、データセットに最大数の列を設定する必要があるため、データセットに対して2回のパスが必要です。

この特定の問題については、私は実際にこのように、代わりにデータフレームまたはデータセットのRDDSとなるだろう：

val data = spark.read.textFile("file.csv").rdd 

val rdd = data.map(s => (s, s.split(",").length)).cache 
val maxColumns = rdd.map(_._2).max() 

val x = rdd 
    .map(row => { 
    val rowData = row._1.split(",") 
    val extraColumns = Array.ofDim[String](maxColumns - rowData.length) 
    Row((rowData ++ extraColumns).toList:_*) 
    }) 

希望に役立つこと:)

Answer 3

をコラムdataTypesと列の数が分かっている場合を定義し、csvファイルをdataframeと読んでいる間にschemaを適用することができます。私はstringType

val schema = StructType(Seq(
    StructField("col1", StringType, true), 
    StructField("col2", StringType, true), 
    StructField("col3", StringType, true), 
    StructField("col4", StringType, true), 
    StructField("col5", StringType, true))) 

val csvDF : DataFrame = sqlContext.read 
    .option("header", "false") 
    .option("delimiter", ",") 
    .option("inferSchema", "false") 
    .schema(schema) 
    .csv("file.csv") 

として、すべての5つの列を定義しているの下にはあなたはそれが各行に、すべての5つの列を追加することが可能である

+----+----+----+----+----+ 
|col1|col2|col3|col4|col5| 
+----+----+----+----+----+ 
|A1 |B1 |C1 |null|null| 
|A2 |B2 |C2 |D1 |null| 
|A3 |B3 |C3 |D2 |E1 | 
|A4 |B4 |C4 |D3 |null| 
|A5 |B5 |C5 |null|E2 | 
+----+----+----+----+----+