1

下面是输入文件（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. 读取此多行输入并将其以表格格式加载到数据库的有效方法是什么

    感谢有关此的任何帮助/指示。谢谢。

4

0 回答 0