java - Sparkで複数行の入力フォーマットを1つのレコードに読み込む最良の方法は何ですか?

Question

以下は、入力ファイル (csv) のようなものです。

Carrier_create_date,Message,REF_SHEET_CREATEDATE,7/1/2008 Carrier_create_time,Message,REF_SHEET_CREATETIME,8:53:57 Carrier_campaign,Analog,REF_SHEET_CAMPAIGN,25 Carrier_run_no,Analog,REF_SHEET_RUNNO,7

以下は、各行に含まれる列のリストです: (Carrier_create_date、Carrier_create_time、Carrier_campaign、Carrier_run_no)

データフレームとしての望ましい出力:

2008 年 7 月 1 日、8:53:57、25、7

基本的に、入力ファイルには、各行に列名と値があります。

私がこれまでに試したことは次のとおりです。

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.{SparkContext, SparkConf}

object coater4CR {

  // Define the application Name
  val AppName: String = "coater4CR"

  // Set the logging level.ERROR)
  Logger.getLogger("org.apache").setLevel(Level.ERROR)

  def main(args: Array[String]): Unit = {

    // define the input parmeters
    val input_file = "/Users/gangadharkadam/myapps/NlrPraxair/src/main/resources/NLR_Praxair›2008›3QTR2008›Coater_4›C025007.csv"

    // Create the Spark configuration and the spark context
    println("Initializing the Spark Context...")

    val conf = new SparkConf().setAppName(AppName).setMaster("local")

    // Define the Spark Context
    val sc = new SparkContext(conf)

    // Read the csv file
    val inputRDD = sc.wholeTextFiles(input_file)
      .flatMap(x => x._2.split(" "))
      .map(x => {
        val rowData = x.split("\n")

        var Carrier_create_date: String = ""
        var Carrier_create_time: String = ""
        var Carrier_campaign: String = ""
        var Carrier_run_no: String = ""

        for (data <- rowData) {
          if (data.trim().startsWith("Carrier_create_date")) {
            Carrier_create_date = data.split(",")(3)
          } else if (data.trim().startsWith("Carrier_create_time")) {
            Carrier_create_time = data.split(",")(3)
          } else if (data.trim().startsWith("Carrier_campaign")) {
            Carrier_campaign = data.split(",")(3)
          } else if (data.trim().startsWith("Carrier_run_no")) {
            Carrier_run_no = data.split(",")(3)
          }
        }
        (Carrier_create_date, Carrier_create_time, Carrier_campaign, Carrier_run_no)
      }).foreach(println)
  }
}

上記のコードを実行すると、上記のコードの問題が発生し ます。以下のように空のリストが表示されます (,,,)

私が変わるとき

Carrier_campaign = data.split(",")(3)

に

Carrier_campaign = data.split(",")(2)

やや近い以下の出力を取得しています (REF_SHEET_CREATEDATE,REF_SHEET_CREATETIME,REF_SHEET_CAMPAIGN,REF_SHEET_RUNNO) (,,,)

上記のコードは、データ行から最後の列位置を選択できませんが、列位置 0、1、2 に対しては機能しています。

だから私の質問は-

1. 上記のコードの何が問題なのですか

2. この複数行の入力を読み取り、表形式でデータベースにロードする効率的な方法は何ですか

   これに関するヘルプ/ポインタに感謝します。ありがとう。