PySpark - 数据框操作答案

【问题标题】：PySpark - Dataframe ManipulationsPySpark - 数据框操作
【发布时间】：2019-11-19 18:08:33
【问题描述】：

我是 Spark 的新手，我尝试了各种方法来了解 Spark。目前，我有一个 CSV，我正在尝试将其解析并操作为我所需的格式。我不了解如何进行旋转并获得输出或通过任何其他方式。我的 CSV 看起来像这样：

AHeader AValue, BHeader BValue, CHeader CValue

现在我尝试构建的 CSV 输出是这样的：

AHeader, AValue
BHeader, BValue
CHeader, CValue

这是我当前的代码：

datafile_csv = "test.csv"

def process_csv(abspath, sparkcontext):
    sqlContext = SQLContext (sparkcontext)
    df = sqlContext.read.load (os.path.join (abspath, datafile_csv),
                               format='com.databricks.spark.csv',
                               inferSchema='true')

    df.registerTempTable("currency")
    print "Dataframe:"
    display(df)
    // Don't know what to do here ????
    reshaped_df = df.groupby('_c0')
    display(reshaped_df)

if __name__ == "__main__":

    abspath = os.path.abspath(os.path.dirname(__file__))
    conf = (SparkConf ()
            . setMaster("local[20]")
            . setAppName("Currency Parser")
            . set("spark.executor.memory", "2g"))
    sc = SparkContext(conf=conf)
    process_csv (abspath, sc)

我不确定如何将此数据帧转换为预期的输出。我是否需要将所有列转置为行，然后对它们执行 SparkSQL？什么是正确的解决方案？

【问题讨论】：

您需要使用自定义行分隔符来解析它，我认为 databricks 不支持。试试pandas read_csv，将lineterminator 定义为逗号，空格作为列分隔符。
@Andrew 你能帮我看看如何用 Pandas 做到这一点吗？不确定，因为我是 Spark 生态系统的新手

标签： python apache-spark pyspark apache-spark-sql

【解决方案1】：

你在这里问了两个问题。第一个问题是正确加载 CSV 的 ETL 问题，最好在 pandas 中完成（由于您的数据结构非常具体），例如：

import pandas as pd
from pyspark.sql import SparkSession
from io import StringIO

spark = SparkSession.builder.getOrCreate()
TESTDATA = StringIO("""AHeader AValue, BHeader BValue, CHeader CValue""")

pandas_df = pd.read_csv(TESTDATA,  # replace with path to your csv
                        delim_whitespace=True,
                        lineterminator=",",
                        header=None,
                        names=['col1', 'col2'])
spark_df = spark.createDataFrame(pandas_df)
spark_df.show()

+-------+------+
|   col1|  col2|
+-------+------+
|AHeader|AValue|
|BHeader|BValue|
|CHeader|CValue|
+-------+------+

您的第二个问题是关于spark 的支点。虽然pandas.read_csv() 将其放入您要求的形状，但如果您需要进一步重塑，请看这里：http://spark.apache.org/docs/2.4.3/api/python/pyspark.sql.html?highlight=pivot#pyspark.sql.GroupedData.pivot

【讨论】：