データフレーム

Question

は、私がこのような構造で大きなデータフレーム（1.2ギガバイトは、多かれ少なかれ）はスパークに複数のものに1列を

変革：行の数は300.000で、「テキスト」フィールドが文字列である

 
+---------+--------------+------------------------------------------------------------------------------------------------------+ 
| country | date_data |             text             | 
+---------+--------------+------------------------------------------------------------------------------------------------------+ 
| "EEUU" | "2016-10-03" | "T_D: QQWE\nT_NAME: name_1\nT_IN: ind_1\nT_C: c1ws12\nT_ADD: Sec_1_P\n ...........\nT_R: 45ee"  | 
| "EEUU" | "2016-10-03" | "T_D: QQAA\nT_NAME: name_2\nT_IN: ind_2\nT_C: c1ws12\nT_ADD: Sec_1_P\n ...........\nT_R: 46ee"  | 
| .  | .   | .                         | 
| .  | .   | .                         | 
| "EEUU" | "2016-10-03" | "T_D: QQWE\nT_NAME: name_300000\nT_IN: ind_65\nT_C: c1ws12\nT_ADD: Sec_1_P\n ...........\nT_R: 47aa" | 
+---------+--------------+------------------------------------------------------------------------------------------------------+ 

を約5000文字です。

私は、この新しいフィールドにフィールド「テキスト」を分離したいと思います：

 
+---------+------------+------+-------------+--------+--------+---------+--------+------+ 
| country | date_data | t_d | t_name | t_in | t_c | t_add | ...... | t_r | 
+---------+------------+------+-------------+--------+--------+---------+--------+------+ 
| EEUU | 2016-10-03 | QQWE | name_1  | ind_1 | c1ws12 | Sec_1_P | ...... | 45ee | 
| EEUU | 2016-10-03 | QQAA | name_2  | ind_2 | c1ws12 | Sec_1_P | ...... | 45ee | 
| .  | .   | . | .   | .  | .  | .  | .  |  | 
| .  | .   | . | .   | .  | .  | .  | .  |  | 
| .  | .   | . | .   | .  | .  | .  | .  |  | 
| EEUU | 2016-10-03 | QQWE | name_300000 | ind_65 | c1ws12 | Sec_1_P | ...... | 47aa | 
+---------+------------+------+-------------+--------+--------+---------+--------+------+ 

は現在、正規表現を使用して - 私はこの問題を解決すること。まず、私は、通常expresionsを記述したテキスト（合計で90個の正規表現）からの個々のフィールドを抽出する関数を作成する：

val D_text = "((?<=T_D:).*?(?=\\\\n))".r 
val NAME_text = "((?<=nT_NAME:).*?(?=\\\\n))".r 
val IN_text = "((?<=T_IN:).*?(?=\\\\n))".r 
val C_text = "((?<=T_C:).*?(?=\\\\n))".r 
val ADD_text = "((?<=T_ADD:).*?(?=\\\\n))".r 
     . 
     . 
     . 
     . 
val R_text = "((?<=T_R:).*?(?=\\\\n))".r 

//UDF function: 
def getFirst(pattern2: scala.util.matching.Regex) = udf(
      (url: String) => pattern2.findFirstIn(url) match { 
       case Some(texst_new) => texst_new 
       case None => "NULL" 
       case null => "NULL" 
      } 
    ) 

そして、私は新たなデータフレーム（tbl_separate_fields）を作成して機能を適用した結果としてテキストから新しいフィールドをすべて抽出する正規表現。

val tbl_separate_fields = hiveDF.select(
      hiveDF("country"), 
      hiveDF("date_data"), 
      getFirst(D_text)(hiveDF("texst")).alias("t_d"), 
      getFirst(NAME_text)(hiveDF("texst")).alias("t_name"), 
      getFirst(IN_text)(hiveDF("texst")).alias("t_in"), 
      getFirst(C_text)(hiveDF("texst")).alias("t_c"), 
      getFirst(ADD_text)(hiveDF("texst")).alias("t_add"), 
          . 
          . 
          . 
          . 

     getFirst(R_text)(hiveDF("texst")).alias("t_r") 

     ) 

は最後に、私はハイブテーブルにこのデータフレームを挿入します。私は最適化し、実行時間を短縮したいので

tbl_separate_fields.registerTempTable("tbl_separate_fields") 
hiveContext.sql("INSERT INTO TABLE TABLE_INSERT PARTITION (date_data) SELECT * FROM tbl_separate_fields") 

このソリューションは、全体のデータフレームのための1時間持続します。解決策はありますか？

ハドープ2.7.1とApache-Spark 1.5.1です。スパークの設定は：

val conf = new SparkConf().set("spark.storage.memoryFraction", "0.1") 
val sc = new SparkContext(conf) 
val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc) 

ありがとうございます。

EDITデータ：

 
+---------+--------------+------------------------------------------------------------------------------------------------------+ 
| country | date_data |             text             | 
+---------+--------------+------------------------------------------------------------------------------------------------------+ 
| "EEUU" | "2016-10-03" | "T_D: QQWE\nT_NAME: name_1\nT_IN: ind_1\nT_C: c1ws12\nT_ADD: Sec_1_P\n ...........\nT_R: 45ee"  | 
| "EEUU" | "2016-10-03" | "T_NAME: name_2\nT_D: QQAA\nT_IN: ind_2\nT_C: c1ws12 ...........\nT_R: 46ee"       | 
| .  | .   | .                         | 
| .  | .   | .                         | 
| "EEUU" | "2016-10-03" | "T_NAME: name_300000\nT_ADD: Sec_1_P\nT_IN: ind_65\nT_C: c1ws12\n ...........\nT_R: 47aa"   | 
+---------+--------------+------------------------------------------------------------------------------------------------------+ 

Answer 1

正規表現を使用して、この場合には、遅くても壊れやすいです。

あなたはすべてのレコードが同じ構造を持っていることがわかっている場合は、（任意の数の列のための）すべての「テキスト」の値が「部品」の同じ数と順を持って、次のコードが動作するということ、つまり、主にorg.apache.spark.sql.functionsでsplit機能を利用して：上記の仮定が真でない場合は別の方法として

import org.apache.spark.sql.functions._ 

// first - split "text" column values into Arrays 
val textAsArray: DataFrame = inputDF 
    .withColumn("as_array", split(col("text"), "\n")) 
    .drop("text") 
    .cache() 

// get a sample (first row) to get column names, can be skipped if you want to hard-code them: 
val sampleText = textAsArray.first().getAs[mutable.WrappedArray[String]]("as_array").toArray 
val columnNames: Array[(String, Int)] = sampleText.map(_.split(": ")(0)).zipWithIndex 

// add Column per columnName with the right value and drop the no-longer-needed as_array column 
val withValueColumns: DataFrame = columnNames.foldLeft(textAsArray) { 
    case (df, (colName, index)) => df.withColumn(colName, split(col("as_array").getItem(index), ": ").getItem(1)) 
}.drop("as_array") 

withValueColumns.show() 
// for the sample data I created, 
// with just 4 "parts" in "text" column, this prints: 
// +-------+----------+----+------+-----+------+ 
// |country| date_data| T_D|T_NAME| T_IN| T_C| 
// +-------+----------+----+------+-----+------+ 
// | EEUU|2016-10-03|QQWE|name_1|ind_1|c1ws12| 
// | EEUU|2016-10-03|QQAA|name_2|ind_2|c1ws12| 
// +-------+----------+----+------+-----+------+ 

を、あなたはMapにテキスト列を変換し、UDFを使用して、類似したreduceLeftオペを行うことができます目的の列のハードコードされたリストの編集：

import sqlContext.implicits._ 

// sample data: not the same order, not all records have all columns: 
val inputDF: DataFrame = sc.parallelize(Seq(
    ("EEUU", "2016-10-03", "T_D: QQWE\nT_NAME: name_1\nT_IN: ind_1\nT_C: c1ws12"), 
    ("EEUU", "2016-10-03", "T_D: QQAA\nT_IN: ind_2\nT_NAME: name_2") 
)).toDF("country", "date_data", "text") 

// hard-coded list of expected column names: 
val columnNames: Seq[String] = Seq("T_D", "T_NAME", "T_IN", "T_C") 

// UDF to convert text into key-value map 
val asMap = udf[Map[String, String], String] { s => 
    s.split("\n").map(_.split(": ")).map { case Array(k, v) => k -> v }.toMap 
} 


val textAsMap = inputDF.withColumn("textAsMap", asMap(col("text"))).drop("text") 

// for each column name - lookup the value in the map 
val withValueColumns: DataFrame = columnNames.foldLeft(textAsMap) { 
    case (df, colName) => df.withColumn(colName, col("textAsMap").getItem(colName)) 
}.drop("textAsMap") 

withValueColumns.show() 
// prints: 
// +-------+----------+----+------+-----+------+ 
// |country| date_data| T_D|T_NAME| T_IN| T_C| 
// +-------+----------+----+------+-----+------+ 
// | EEUU|2016-10-03|QQWE|name_1|ind_1|c1ws12| 
// | EEUU|2016-10-03|QQAA|name_2|ind_2| null| 
// +-------+----------+----+------+-----+------+