Pypark数据帧操作

Question

我是新的火花和尝试各种事情，以了解火花。目前，我有一个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？正确的解决方法是什么？你知道吗

Answer 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