reduceByKey PySpark 中的列表列表

reduceByKey a list of lists in PySpark

我是 pyspark 的新手,到目前为止,当您习惯使用像 pandas 这样的库时,很难理解它的特殊工作方式。但这似乎是大数据的出路。

对于我目前的 ETL 工作,我有以下要素:

这是我的 rdd:

[
    [
    ('SMSG', 'BKT'), ('SQNR', '00000004'), ('STNQ', '06'), ('TRNN', '000001'), ('SMSG', 'BKS'), ('SQNR', '00000005'), ('STNQ', '24'), ('DAIS', '171231'), ('TRNN', '000001'), ....
    ], 
    [
    ('SMSG', 'BKT'), ('SQNR', '00000024'), ('STNQ', '06'), ('TRNN', '000002'), ('NRID', '  '), ('TREC', '020'), ('TRNN', '000002'), ('NRID', '  '), ('TACN', '001'), ('CARF', '          '), ... 
    ],
    ...
]

行数据是一个固定大小的文本文件。

我现在要做的是对列表的每个单元格进行 groupByKey。

最终结果应该是:

[
    [
    ('SMSG_1', 'BKT'),('SMSG_2','BKS'),('SQNR_1', '00000004'),('SQNR_2', '00000005'),('STNQ_1','06'),('STNQ_2','24'),('TRNN', '000001'),()('DAIS', '171231'),...
    ],
    [
    ('SMSG', 'BKT'),('SQNR', '00000024'),('STNQ','06'),('TRNN', '000002'),('NRID', '  '), ('TREC', '020'), ('TACN', '001'), ('CARF', '          '),...
    ],
    ...
]

基本规则如下:

1- 如果键相同且值也相同,则删除重复项。

2- 如果键相同而值不同,重命名列并添加后缀为“_Number”,其中 Number 可以替换为该键的迭代次数。

我的代码开始如下:

def addBKT():
...
def prepareTrans():
...
if __name__ == '__main__':
    input_folder = '/Users/admin/Documents/Training/FR20180101HOT' 
    rdd = sc.wholeTextFiles(input_folder).map(lambda x: x[1].split("BKT"))
    rdd = rdd.flatMap(prepareTrans).map(addBKT).map(lambda x: x.split("\n")).map(hot_to_flat_file_v2)
    print(rdd.take(1))

打印给我(如前所述)以下元组列表列表。我只拿了 1 个子列表,但完整的 rdd 有大约 2000 个元组子列表:

[
    [
    ('SMSG', 'BKT'), ('SQNR', '00000004'), ('STNQ', '06'), ('TRNN', '000001'), ('SMSG', 'BKS'), ('SQNR', '00000005'), ('STNQ', '24'), ('DAIS', '171231'), ('TRNN', '000001'), ....
    ]
]

我尝试首先减少嵌套列表如下:

rdd = rdd.flatMap(lambda x:x).reduceByKey(list)

我期待得到一个没有重复的新列表列表,并且对于具有不同值的元组,将它们全部分组在同一个键下。但是,我做不到。

作为第二步,我计划将具有多个值的元组转换为新的元组对,尽可能多地在分组元组中获得值:即('Key'、['Value1'、'Value2']) 变成 ('Key_1', 'Value1'),('Key_2', 'Value2')

最后,所有这些转换的输出是将最终的 RDD 转换为 DataFrame 并以 parquet 格式存储。

我真的希望过去有人做过类似的事情。我花了很多时间尝试这样做,但我无法做到,也无法在网上找到任何示例。

感谢您的帮助。

由于您是 spark 的新手,您可能不知道 Spark Dataframe。与RDD相比,Dataframe是高级概念。在这里,我使用 Pyspark Dataframe 解决了您的问题。看看这个,不要犹豫学习 spark Dataframe。

rdd1 = sc.parallelize([("SMSG", "BKT"), ("SMSG", "BKT"), ("SMSG", "BKS"), ('SQNR', '00000004'), ('SQNR', '00000005') ])
rddToDF = rdd1.toDF(["C1", "C2"])
+----+--------+
|  C1|      C2|
+----+--------+
|SMSG|     BKT|
|SMSG|     BKT|
|SMSG|     BKS|
|SQNR|00000004|
|SQNR|00000005|
+----+--------+

DfRmDup = rddToDF.drop_duplicates() #Removing duplicates from Dataframe
DfRmDup.show()
+----+--------+
|  C1|      C2|
+----+--------+
|SQNR|00000004|
|SMSG|     BKT|
|SQNR|00000005|
|SMSG|     BKS|
+----+--------+

rank = DfRmDup.withColumn("rank", dense_rank().over(Window.partitionBy("C1").orderBy(asc("C2"))))
rank.show()
+----+--------+----+
|  C1|      C2|rank|
+----+--------+----+
|SQNR|00000004|   1|
|SQNR|00000005|   2|
|SMSG|     BKS|   1|
|SMSG|     BKT|   2|
+----+--------+----+

rank.withColumn("C1", concat(col("C1"), lit("_"), col("rank"))).drop("rank").show()
+------+--------+
|    C1|      C2|
+------+--------+
|SQNR_1|00000004|
|SQNR_2|00000005|
|SMSG_1|     BKS|
|SMSG_2|     BKT|
+------+--------+

#Converting back to RDD
rank.withColumn("C1", concat(col("C1"), lit("_"), col("rank"))).drop("rank").rdd.map(lambda x: (x[0],x[1])).collect()

[('SQNR_1', '00000004'),
 ('SQNR_2', '00000005'),
 ('SMSG_1', 'BKS'),
 ('SMSG_2', 'BKT')]

非常感谢link,我遵循了提供的解决方案。数据框已成功创建,这很棒。

    input_folder = '/Users/admin/Documents/Training/FR20180101HOT' 
    rdd_split = sc.wholeTextFiles(input_folder).map(lambda x: x[1].split("BKT"))
    rdd_trans = rdd_split.flatMap(prepareTrans).map(addBKT).map(lambda x: x.split("\n")).map(hot_to_flat_file_v2)
    #rdd_group = rdd_trans.map(lambda x : x[i] for i in range(len(x))).reduceByKey(lambda x, y: str(x) + ','+ str(y))   
    df = spark.read.options(inferSchema="true").csv(rdd_trans)
    print(df.show(1)) 

印刷品向我展示了类似的东西:

+--------+-------+--------+------------+--------+------+--------+----------+----...

|     _c0|    _c1|     _c2|         _c3|     _c4|   _c5|     _c6|       _c7|     _c8|   _c9|    _c10|   _c11|    _c12|   _c13|    _c14|          _c15|    _c16|                _c17|    _c18|    _c19|    _c20|              _c21|    _c22| _c23|    _c24| _c25|    _c26| _c27|    _c28| _c29|    _c30|                _c31|    _c32|   _c33|    _c34|        _c35|    _c36|  _c37|    _c38|      _c39|    _c40|      _c41|    _c42|              _c43|    _c44| _c45|    _c46|    _c47|    _c48|   _c49|    _c50|        _c51|    _c52| _c53|    _c54|               _c55|    _c56|    _c57|    _c58|          _c59|    _c60|             _c61|    _c62|    _c63|    _c64|                _c65|    _c66|   _c67|    _c68|        _c69|    _c70|  _c71|    _c72|      _c73|    _c74|      _c75|    _c76|              _c77|    _c78| _c79|    _c80|   _c81|    _c82| _c83|      _c84|        _c85|      _c86|   _c87|      _c88|        _c89|      _c90|   _c91|      _c92|        _c93|      _c94|   _c95|    _c96|    _c97|    _c98|  _c99|   _c100|   _c101|   _c102|  _c103|   _c104|       _c105|   _c106| _c107|   _c108|     _c109|   _c110|     _c111|   _c112|             _c113|   _c114|_c115|   _c116|_c117|   _c118|_c119|     _c120|       _c121|     _c122| _c123|     _c124|       _c125|     _c126| _c127|     _c128|       _c129|     _c130|  _c131|   _c132|_c133|   _c134| _c135|   _c136|   _c137|   _c138|  _c139|   _c140|       _c141|   _c142| _c143|   _c144|     _c145|   _c146|     _c147|   _c148|             _c149|   _c150|_c151|   _c152|_c153|   _c154|_c155|     _c156|       _c157|     _c158| _c159|     _c160|       _c161|     _c162|_c163|     _c164|       _c165|     _c166|  _c167|   _c168|_c169|   _c170| _c171|   _c172|   _c173|   _c174|  _c175|   _c176|       _c177|   _c178| _c179|   _c180|     _c181|   _c182|     _c183|   _c184|             _c185|   _c186|_c187|   _c188|_c189|   _c190|_c191|     _c192|       _c193|     _c194|  _c195|     _c196|       _c197|     _c198| _c199|     _c200|       _c201|     _c202|  _c203|   _c204|_c205|   _c206| _c207|   _c208|   _c209|   _c210|  _c211|   _c212|       _c213|   _c214| _c215|   _c216|     _c217|   _c218|     _c219|   _c220|             _c221|   _c222|_c223|   _c224|_c225|   _c226|_c227|     _c228|       _c229|     _c230| _c231|     _c232|       _c233|     _c234| _c235|     _c236|       _c237|     _c238|  _c239|   _c240|_c241|   _c242| _c243|   _c244|   _c245|   _c246|  _c247|   _c248|       _c249|   _c250| _c251|   _c252|     _c253|   _c254|     _c255|   _c256|             _c257|   _c258|_c259|   _c260|_c261|   _c262|_c263|     _c264|       _c265|     _c266| _c267|     _c268|       _c269|     _c270|_c271|     _c272|       _c273|     _c274|_c275|   _c276|_c277|   _c278| _c279|   _c280|   _c281|   _c282|  _c283|   _c284|       _c285|   _c286| _c287|   _c288|     _c289|   _c290|     _c291|   _c292|             _c293|   _c294|_c295|   _c296|  _c297|   _c298|     _c299|   _c300|    _c301|   _c302|_c303|   _c304|     _c305|   _c306|    _c307|   _c308|_c309|   _c310|    _c311|   _c312|_c313|   _c314|_c315|   _c316|_c317|   _c318|               _c319|   _c320|   _c321|   _c322|  _c323|   _c324|       _c325|   _c326| _c327|   _c328|     _c329|   _c330|     _c331|   _c332|             _c333|   _c334|_c335|   _c336|             _c337|   _c338|  _c339|   _c340|      _c341|   _c342|       _c343|   _c344|               _c345|   _c346|              _c347|   _c348|  _c349|   _c350|       _c351|   _c352| _c353|   _c354|     _c355|   _c356|     _c357|   _c358|             _c359|   _c360|_c361|   _c362|_c363|   _c364|_c365|   _c366|    _c367|   _c368|    _c369|   _c370|    _c371|   _c372|    _c373|   _c374|  _c375|   _c376|_c377|   _c378|    _c379|   _c380| _c381|   _c382|    _c383|   _c384|    _c385|   _c386| _c387|   _c388|  _c389|   _c390|              _c391|   _c392|               _c393|   _c394|  _c395|   _c396|         _c397|   _c398|  _c399|   _c400|       _c401|   _c402| _c403|   _c404|     _c405|   _c406|     _c407|   _c408|             _c409|   _c410|_c411|   _c412|_c413|   _c414|_c415|   _c416|    _c417|   _c418|    _c419|   _c420|    _c421|   _c422|    _c423|   _c424|  _c425|   _c426|_c427|   _c428|    _c429|   _c430| _c431|   _c432|    _c433|   _c434|    _c435|   _c436| _c437|   _c438|  _c439|   _c440|              _c441|   _c442|               _c443|   _c444|  _c445|   _c446|         _c447|   _c448|  _c449|   _c450|       _c451|   _c452| _c453|   _c454|     _c455|   _c456|     _c457|   _c458|             _c459|   _c460|_c461|   _c462|_c463|   _c464|_c465|   _c466|    _c467|   _c468|    _c469|   _c470|    _c471|   _c472|    _c473|   _c474|  _c475|   _c476|_c477|   _c478|    _c479|   _c480| _c481|   _c482|    _c483|   _c484|    _c485|   _c486| _c487|   _c488|  _c489|   _c490|              _c491|   _c492|               _c493|   _c494|  _c495|   _c496|         _c497|   _c498|  _c499|   _c500|       _c501|   _c502| _c503|   _c504|     _c505|   _c506|     _c507|   _c508|             _c509|   _c510|_c511|   _c512|_c513|   _c514|_c515|   _c516|    _c517|   _c518|    _c519|   _c520|    _c521|   _c522|    _c523|   _c524|  _c525|   _c526|_c527|   _c528|    _c529|   _c530| _c531|   _c532|    _c533|   _c534|    _c535|   _c536| _c537|   _c538|  _c539|   _c540|              _c541|   _c542|               _c543|   _c544|  _c545|   _c546|         _c547|   _c548|  _c549|   _c550|       _c551|   _c552| _c553|   _c554|     _c555|   _c556|     _c557|   _c558|             _c559|   _c560|_c561|   _c562|           _c563|   _c564|_c565|   _c566|           _c567|   _c568|           _c569|   _c570|   _c571|   _c572|_c573|   _c574|     _c575|   _c576|_c577|   _c578|_c579|   _c580|       _c581|   _c582|               _c583|   _c584|  _c585|   _c586|       _c587|   _c588| _c589|   _c590|     _c591|   _c592|     _c593|   _c594|             _c595|   _c596|_c597|   _c598|               _c599|   _c600|               _c601|   _c602|      _c603|   _c604|  _c605|   _c606|       _c607|   _c608|  _c609|   _c610|       _c611|   _c612| _c613|   _c614|     _c615|   _c616|     _c617|   _c618|             _c619|   _c620|_c621|   _c622|_c623|   _c624|               _c625|   _c626|               _c627|   _c628|  _c629|   _c630|       _c631|   _c632| _c633|   _c634|     _c635|   _c636|     _c637|   _c638|             _c639|   _c640|_c641|   _c642|_c643|   _c644|               _c645|   _c646|       _c647|   _c648|  _c649|   _c650|       _c651|   _c652| _c653|   _c654|     _c655|   _c656|     _c657|   _c658|             _c659|   _c660|_c661|   _c662|_c663|   _c664|               _c665|   _c666|       _c667|   _c668|  _c669|   _c670|       _c671|   _c672| _c673|   _c674|     _c675|   _c676|     _c677|   _c678|         _c679|   _c680|   _c681|   _c682|               _c683|   _c684|   _c685|   _c686| _c687|   _c688|     _c689|   _c690|             _c691|   _c692|     _c693|   _c694|   _c695|   _c696|_c697|   _c698|               _c699|   _c700|    _c701|
+--------+-------+--------+------------+--------+------+--------+----------+-------...

|[('SMSG'| 'BKT')| ('SQNR'| '00000004')| ('STNQ'| '06')| ('TRNN'| '000001')| ('NRID'| '  ')| ('TREC'| '020')| ('TACN'| '001')| ('CARF'| '          ')| ('CSTF'| '               ...| ('RPSI'| 'SABR')| ('ESAC'| '              ')| ('DISI'| ' ')| ('NRMI'| ' ')| ('NRCT'| ' ')| ('AREI'| ' ')| ('RESD'| '               ...| ('SMSG'| 'BKS')| ('SQNR'| '00000005')| ('STNQ'| '24')| ('DAIS'| '171231')| ('TRNN'| '000001')| ('TDNR'| '0015131574285 ')| ('CDGT'| '2')| ('CPUI'| 'FFFF')| ('CJCP'| '   ')| ('AGTN'| '20212146')| ('RFIC'| ' ')| ('TOUR'| '               ')| ('TRNC'| 'TKTT')| ('TODC'| 'CDGCDG    ')| ('PNRR'| 'IKQOWZ/AA    ')| ('TIIS'| '0000')| ('RESD'| '               ...| ('SMSG'| 'BKS')| ('SQNR'| '00000006')| ('STNQ'| '30')| ('DAIS'| '171231')| ('TRNN'| '000001')| ('TDNR'| '0015131574285 ')| ('CDGT'| '2')| ('COBL'| 225.0)| ('NTFA'| 0.0)| ('TMFT_1'| 'YR      ')| ('TMFA_1'| 300.0)| ('TMFT_2'| 'FR      ')| ('TMFA_2'| 20.81)| ('TMFT_3'| 'QX      ')| ('TMFA_3'| 27.91)| ('TDAM'| 712.92)| ('RESD'| '  ')| ('CUTP'| 'EUR2')| ('SMSG'| 'BKS')| ('SQNR'| '00000007')| ('STNQ'| '30')| ('DAIS'| '171231')| ('TRNN'| '000001')| ('TDNR'| '0015131574285 ')| ('CDGT'| '2')| ('COBL'| 0.0)| ('NTFA'| 0.0)| ('TMFT_1'| 'IZ      ')| ('TMFA_1'| 4.51)| ('TMFT_2'| 'YC      ')| ('TMFA_2'| 9.22)| ('TMFT_3'| 'XY      ')| ('TMFA_3'| 11.74)| ('TDAM'| 0.0)| ('RESD'| '  ')| ('CUTP'| 'EUR2')| ('SMSG'| 'BKS')| ('SQNR'| '00000008')| ('STNQ'| '30')| ('DAIS'| '171231')| ('TRNN'| '000001')| ('TDNR'| '0015131574285 ')| ('CDGT'| '2')| ('COBL'| 0.0)| ('NTFA'| 0.0)| ('TMFT_1'| 'XA      ')| ('TMFA_1'| 6.64)| ('TMFT_2'| 'AY      ')| ('TMFA_2'| 9.4)| ('TMFT_3'| 'WD      ')| ('TMFA_3'| 29.33)| ('TDAM'| 0.0)| ('RESD'| '  ')| ('CUTP'| 'EUR2')| ('SMSG'| 'BKS')| ('SQNR'| '00000009')| ('STNQ'| '30')| ('DAIS'| '171231')| ('TRNN'| '000001')| ('TDNR'| '0015131574285 ')| ('CDGT'| '2')| ('COBL'| 0.0)| ('NTFA'| 0.0)| ('TMFT_1'| 'EK      ')| ('TMFA_1'| 18.89)| ('TMFT_2'| 'EL      ')| ('TMFA_2'| 4.19)| ('TMFT_3'| 'HG      ')| ('TMFA_3'| 16.76)| ('TDAM'| 0.0)| ('RESD'| '  ')| ('CUTP'| 'EUR2')| ('SMSG'| 'BKS')| ('SQNR'| '00000010')| ('STNQ'| '30')| ('DAIS'| '171231')| ('TRNN'| '000001')| ('TDNR'| '0015131574285 ')| ('CDGT'| '2')| ('COBL'| 0.0)| ('NTFA'| 0.0)| ('TMFT_1'| 'JT      ')| ('TMFA_1'| 2.52)| ('TMFT_2'| 'UC      ')| ('TMFA_2'| 6.72)| ('TMFT_3'| 'QK      ')| ('TMFA_3'| 16.76)| ('TDAM'| 0.0)| ('RESD'| '  ')| ('CUTP'| 'EUR2')| ('SMSG'| 'BKS')| ('SQNR'| '00000011')| ('STNQ'| '30')| ('DAIS'| '171231')| ('TRNN'| '000001')| ('TDNR'| '0015131574285 ')| ('CDGT'| '2')| ('COBL'| 0.0)| ('NTFA'| 0.0)| ('TMFT_1'| 'XF      ')| ('TMFA_1'| 2.52)| ('TMFT_2'| 'XFCLT3  ')| ('TMFA_2'| 0.0)| ('TMFT_3'| '        ')| ('TMFA_3'| 0.0)| ('TDAM'| 0.0)| ('RESD'| '  ')| ('CUTP'| 'EUR2')| ('SMSG'| 'BKS')| ('SQNR'| '00000012')| ('STNQ'| '39')| ('DAIS'| '171231')| ('TRNN'| '000001')| ('TDNR'| '0015131574285 ')| ('CDGT'| '2')| ('STAT'| 'I  ')| ('COTP'| '      ')| ('CORT'| '00000')| ('COAM'| 0.0)| ('SPTP'| '      ')| ('SPRT'| '00000')| ('SPAM'| 0.0)| ('EFRT'| '00000')| ('EFCO'| 0.0)| ('APBC'| 0.0)| ('RDII'| ' ')| ('RESD'| '               ...| ('CUTP'| 'EUR2')| ('SMSG'| 'BKS')| ('SQNR'| '00000013')| ('STNQ'| '46')| ('DAIS'| '171231')| ('TRNN'| '000001')| ('TDNR'| '0015131574285 ')| ('CDGT'| '2')| ('ORIT'| '              ')| ('ORIL'| '   ')| ('ORID'| '       ')| ('ORIA'| '00000000')| ('ENRS'| 'NONREF/RESTRICT...| ('RESD'| '               ')| ('SMSG'| 'BKI')| ('SQNR'| '00000014')| ('STNQ'| '63')| ('DAIS'| '171231')| ('TRNN'| '000001')| ('TDNR'| '0015131574285 ')| ('CDGT'| '2')| ('SEGI'| '1')| ('STPO'| 'X')| ('NBDA'| '22APR')| ('NADA'| '22APR')| ('ORAC'| 'CDG  ')| ('DSTC'| 'MIA  ')| ('CARR'| 'AA ')| ('CABI'| ' ')| ('FTNR'| '  63 ')| ('RBKD'| 'O ')| ('FTDA'| '22APR')| ('FTDT'| '1155 ')| ('FBST'| 'OK')| ('FBAL'| '1PC')| ('FBTD'| 'OLN0DMN3       ')| ('FFRF'| '               ...| ('FCPT'| '   ')| ('RESD'| '          ')| ('SMSG'| 'BKI')| ('SQNR'| '00000015')| ('STNQ'| '63')| ('DAIS'| '171231')| ('TRNN'| '000001')| ('TDNR'| '0015131574285 ')| ('CDGT'| '2')| ('SEGI'| '2')| ('STPO'| 'O')| ('NBDA'| '22APR')| ('NADA'| '22APR')| ('ORAC'| 'MIA  ')| ('DSTC'| 'MBJ  ')| ('CARR'| 'AA ')| ('CABI'| ' ')| ('FTNR'| '1515 ')| ('RBKD'| 'O ')| ('FTDA'| '22APR')| ('FTDT'| '1801 ')| ('FBST'| 'OK')| ('FBAL'| '1PC')| ('FBTD'| 'OLN0DMN3       ')| ('FFRF'| '               ...| ('FCPT'| '   ')| ('RESD'| '          ')| ('SMSG'| 'BKI')| ('SQNR'| '00000016')| ('STNQ'| '63')| ('DAIS'| '171231')| ('TRNN'| '000001')| ('TDNR'| '0015131574285 ')| ('CDGT'| '2')| ('SEGI'| '3')| ('STPO'| 'X')| ('NBDA'| '29APR')| ('NADA'| '29APR')| ('ORAC'| 'MBJ  ')| ('DSTC'| 'CLT  ')| ('CARR'| 'AA ')| ('CABI'| ' ')| ('FTNR'| ' 844 ')| ('RBKD'| 'O ')| ('FTDA'| '29APR')| ('FTDT'| '1059 ')| ('FBST'| 'OK')| ('FBAL'| '1PC')| ('FBTD'| 'OLN0DMN3       ')| ('FFRF'| '               ...| ('FCPT'| '   ')| ('RESD'| '          ')| ('SMSG'| 'BKI')| ('SQNR'| '00000017')| ('STNQ'| '63')| ('DAIS'| '171231')| ('TRNN'| '000001')| ('TDNR'| '0015131574285 ')| ('CDGT'| '2')| ('SEGI'| '4')| ('STPO'| ' ')| ('NBDA'| '29APR')| ('NADA'| '29APR')| ('ORAC'| 'CLT  ')| ('DSTC'| 'CDG  ')| ('CARR'| 'AA ')| ('CABI'| ' ')| ('FTNR'| ' 786 ')| ('RBKD'| 'O ')| ('FTDA'| '29APR')| ('FTDT'| '1630 ')| ('FBST'| 'OK')| ('FBAL'| '1PC')| ('FBTD'| 'OLN0DMN3       ')| ('FFRF'| '               ...| ('FCPT'| '   ')| ('RESD'| '          ')| ('SMSG'| 'BAR')| ('SQNR'| '00000018')| ('STNQ'| '64')| ('DAIS'| '171231')| ('TRNN'| '000001')| ('TDNR'| '0015131574285 ')| ('CDGT'| '2')| ('FARE'| 'EUR   225.00')| ('TKMI'| '/')| ('EQFR'| '            ')| ('TOTL'| 'EUR   712.92')| ('SASI'| '0011')| ('FCMI'| '0')| ('BAID'| '      ')| ('BEOT'| ' ')| ('FCPI'| '0')| ('AENT'| '        ')| ('RESD'| '               ...| ('SMSG'| 'BAR')| ('SQNR'| '00000019')| ('STNQ'| '65')| ('DAIS'| '171231')| ('TRNN'| '000001')| ('TDNR'| '0015131574285 ')| ('CDGT'| '2')| ('PXNM'| '               ...| ('PXDA'| '               ...| ('DOBR'| '02APR68')| ('PXTP'| '   ')| ('RESD'| '        ')| ('SMSG'| 'BAR')| ('SQNR'| '00000020')| ('STNQ'| '66')| ('DAIS'| '171231')| ('TRNN'| '000001')| ('TDNR'| '0015131574285 ')| ('CDGT'| '2')| ('FPSN'| '1')| ('FPIN'| 'AA132193       ...| ('RESD'| '               ...| ('SMSG'| 'BKF')| ('SQNR'| '00000021')| ('STNQ'| '81')| ('DAIS'| '171231')| ('TRNN'| '000001')| ('TDNR'| '0015131574285 ')| ('CDGT'| '2')| ('FRCS'| '1')| ('FRCA'| 'PAR AA X/MIA AA...| ('RESD'| '        ')| ('SMSG'| 'BKF')| ('SQNR'| '00000022')| ('STNQ'| '81')| ('DAIS'| '171231')| ('TRNN'| '000001')| ('TDNR'| '0015131574285 ')| ('CDGT'| '2')| ('FRCS'| '2')| ('FRCA'| '1IZ9.22YC11.74X...| ('RESD'| '        ')| ('SMSG'| 'BKP')| ('SQNR'| '00000023')| ('STNQ'| '84')| ('DAIS'| '171231')| ('TRNN'| '000001')| ('FPTP'| 'CA        ')| ('FPAM'| 712.92)| ('FPAC'| '               ...| ('EXDA'| '    ')| ('EXPC'| '  ')| ('APLC'| '      ')| ('INVN'| '              ')| ('INVD'| '000000')| ('REMT'| 712.92)| ('CVVR'| ' ')| ('RESD'| '               ...| ('CUTP'| 'EUR2')]|
+--------+-------+--------+------------+--------+------+--------+----------+-------...

我想我仍然需要遍历每对列,用第一列第一行的值重命名第二列,最后删除每对列的所有第一列。

或者是否可以添加更多选项:

df = spark.read.options(inferSchema="true").csv(rdd_trans)

获得准确正确的数据帧结构?它将避免更多的处理时间(我的目标是比 pandas 版本更快)

与此同时,我尝试做:

df.write.parquet("/Users/admin/Documents/Training/FR20180101HOT.parquet")

但是出现错误:

Py4JJavaError: An error occurred while calling o447851.parquet.
: org.apache.spark.SparkException: Job aborted.
    at org.apache.spark.sql.execution.datasources.FileFormatWriter$.write(FileFormatWriter.scala:196)
    at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelationCommand.run(InsertIntoHadoopFsRelationCommand.scala:159)
    ...
Caused by: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 8220.0 failed 1 times, most recent failure: Lost task 0.0 in stage 8220.0 (TID 12712, localhost, executor driver): org.apache.spark.SparkException: Task failed while writing rows.

...
Caused by: java.lang.OutOfMemoryError: GC overhead limit exceeded
...

由于文本限制,我不能把所有的错误信息都放上去,但它似乎与内存问题有关。

我为 df 做了一个计数:

print(df.count())

15723

这等于我的 pandas 版本(其他 python 代码不使用 pyspark)中的行数,所以它得到正确的行数。但是,在 pandas 中,我可以毫无问题地提取镶木地板。

您可以针对您的情况尝试 regexp_replace。 检查下面的示例案例,

df1.withColumn("c0", regexp_replace("_c0", "[()']", "")).withColumn("c1", regexp_replace("_c1", "\)", "")).show()

+----+---+---+---+
| _c0|_c1| c0| c1|
+----+---+---+---+
|('a'| 2)|  a|  2|
|('b'| 4)|  b|  4|
|('c'| 6)|  c|  6|
+----+---+---+---+