使用 pySpark 读取分号数据的管道

Question

从查找文件中读取文件以及每条记录的位置和国家、状态列 第 1 步：

for line into lines:
    SourceDf = sqlContext.read.format("csv").option("delimiter","|").load(line)
    SourceDf.withColumn("Location",lit("us"))\
    .withColumn("Country",lit("Richmnd"))\
    .withColumn("State",lit("NY"))

第 2 步： 从 DF 上方循环每一列并进行拆分操作，但在 KeyValueDf 中只得到两列。

for col_num in SopurceDf.column:
  InterDF = pyspark.sql.fucntion.split(SourceDf[col_num],":")
  KeyValueDF = SourceDf.withColumn("Column_Name",InterDF.get(0))\
               .withColumn("Column_value",InterDf.get(1))

在第 1 步中：使用管道拆分数据并创建 60 列 在第 2 步中：我想再次用分号分割第 1 步的输出。

谁能帮我请教如何得到预期的结果。 .

*File format :
ABC:"MobileData"|XYZ:"TableData"|ZXC:"MacData"|MNB:"WindowData"
ABC:"value1"    |XYZ:"value2"   |ZXC:"value3" |MNB:"value4"
ABC: "valueA"   |XYZ:"ValueB"   |ZXC:"valueC" |MNB:"valueD"|POI:"valueE"
ABC:"value11"    |XYZ:"value12"   |ZXC:"value13" |MNB:"value14"
ABC:"value1A"    |XYZ:"value2A"   |ZXC:"value3A"

result:
    ABC       | XYZ       |ZXC    |MNB       |POI
    MobileData  TabletData MacData WindowData
    value1      value2     value3  value4    
    valueA      valueB     ValueC  valueD    valueE
    value11     value12    value13 value14    
    value1A      value2A   value3A *
    

Answer 1

我修改了here 给出的解决方案以获得您需要的输出。

我会这样做。

from pyspark.sql import Row
from pyspark.sql.types import *
from pyspark import SparkContext, SQLContext
from pyspark.sql import functions as F
import sys

sc = SparkContext('local')
sqlContext = SQLContext(sc)


input_file = "../data/datafile.csv"

initial_df = sqlContext.read.format("csv").csv(input_file)

initial_df.show(n=100, truncate=False)

inter_df = initial_df.withColumn("array", F.split(initial_df['_c0'], '\|'))

inter_df.show(n=100, truncate=False)


def create_dict(input_string):
    result_list = {}
    for ele in input_string:
        internal_ele = ele.strip()
        internal_ele = internal_ele.split(":")
        internal_ele = [ele.strip() for ele in internal_ele]
        result_list[internal_ele[0]] = internal_ele[1].replace('"', "")
    return result_list

create_dict_udf = F.udf(create_dict, MapType(keyType=StringType(), valueType=StringType()))


inter_df = inter_df.withColumn("dictionary", create_dict_udf(F.col("array")))

inter_df.show(n=100, truncate=False)

keys_df = inter_df.select(F.explode(F.map_keys(F.col("dictionary")))).distinct()
keys = list(map(lambda row: row[0], keys_df.collect()))
key_cols = list(map(lambda f: F.col("dictionary").getItem(f).alias(str(f)), keys))

inter_df.select(key_cols).show()

输出如下。

+------------------------------------------------------------------------+
|_c0                                                                     |
+------------------------------------------------------------------------+
|ABC:"MobileData"|XYZ:"TableData"|ZXC:"MacData"|MNB:"WindowData"         |
|ABC:"value1"    |XYZ:"value2"   |ZXC:"value3" |MNB:"value4"             |
|ABC: "valueA"   |XYZ:"ValueB"   |ZXC:"valueC" |MNB:"valueD"|POI:"valueE"|
|ABC:"value11"    |XYZ:"value12"   |ZXC:"value13" |MNB:"value14"         |
|ABC:"value1A"    |XYZ:"value2A"   |ZXC:"value3A"                        |
+------------------------------------------------------------------------+

+------------------------------------------------------------------------+------------------------------------------------------------------------------+
|_c0                                                                     |array                                                                         |
+------------------------------------------------------------------------+------------------------------------------------------------------------------+
|ABC:"MobileData"|XYZ:"TableData"|ZXC:"MacData"|MNB:"WindowData"         |[ABC:"MobileData", XYZ:"TableData", ZXC:"MacData", MNB:"WindowData"]          |
|ABC:"value1"    |XYZ:"value2"   |ZXC:"value3" |MNB:"value4"             |[ABC:"value1"    , XYZ:"value2"   , ZXC:"value3" , MNB:"value4"]              |
|ABC: "valueA"   |XYZ:"ValueB"   |ZXC:"valueC" |MNB:"valueD"|POI:"valueE"|[ABC: "valueA"   , XYZ:"ValueB"   , ZXC:"valueC" , MNB:"valueD", POI:"valueE"]|
|ABC:"value11"    |XYZ:"value12"   |ZXC:"value13" |MNB:"value14"         |[ABC:"value11"    , XYZ:"value12"   , ZXC:"value13" , MNB:"value14"]          |
|ABC:"value1A"    |XYZ:"value2A"   |ZXC:"value3A"                        |[ABC:"value1A"    , XYZ:"value2A"   , ZXC:"value3A"]                          |
+------------------------------------------------------------------------+------------------------------------------------------------------------------+

+------------------------------------------------------------------------+------------------------------------------------------------------------------+---------------------------------------------------------------------------+
|_c0                                                                     |array                                                                         |dictionary                                                                 |
+------------------------------------------------------------------------+------------------------------------------------------------------------------+---------------------------------------------------------------------------+
|ABC:"MobileData"|XYZ:"TableData"|ZXC:"MacData"|MNB:"WindowData"         |[ABC:"MobileData", XYZ:"TableData", ZXC:"MacData", MNB:"WindowData"]          |[MNB -> WindowData, XYZ -> TableData, ABC -> MobileData, ZXC -> MacData]   |
|ABC:"value1"    |XYZ:"value2"   |ZXC:"value3" |MNB:"value4"             |[ABC:"value1"    , XYZ:"value2"   , ZXC:"value3" , MNB:"value4"]              |[MNB -> value4, XYZ -> value2, ABC -> value1, ZXC -> value3]               |
|ABC: "valueA"   |XYZ:"ValueB"   |ZXC:"valueC" |MNB:"valueD"|POI:"valueE"|[ABC: "valueA"   , XYZ:"ValueB"   , ZXC:"valueC" , MNB:"valueD", POI:"valueE"]|[MNB -> valueD, XYZ -> ValueB, ABC -> valueA, POI -> valueE, ZXC -> valueC]|
|ABC:"value11"    |XYZ:"value12"   |ZXC:"value13" |MNB:"value14"         |[ABC:"value11"    , XYZ:"value12"   , ZXC:"value13" , MNB:"value14"]          |[MNB -> value14, XYZ -> value12, ABC -> value11, ZXC -> value13]           |
|ABC:"value1A"    |XYZ:"value2A"   |ZXC:"value3A"                        |[ABC:"value1A"    , XYZ:"value2A"   , ZXC:"value3A"]                          |[XYZ -> value2A, ABC -> value1A, ZXC -> value3A]                           |
+------------------------------------------------------------------------+------------------------------------------------------------------------------+---------------------------------------------------------------------------+

+------+-------+----------+---------+----------+
|   POI|    ZXC|       MNB|      XYZ|       ABC|
+------+-------+----------+---------+----------+
|  null|MacData|WindowData|TableData|MobileData|
|  null| value3|    value4|   value2|    value1|
|valueE| valueC|    valueD|   ValueB|    valueA|
|  null|value13|   value14|  value12|   value11|
|  null|value3A|      null|  value2A|   value1A|
+------+-------+----------+---------+----------+