如何合并 pyspark 和熊猫数据帧

Question

我有一个非常大的 pyspark 数据帧和一个较小的熊猫数据帧，我读入如下：

df1 = spark.read.csv("/user/me/data1/")
df2 = pd.read_csv("data2.csv")

这两个数据帧都包含标记为“A”和“B”的列。我想创建另一个 pyspark 数据帧，其中只有 df1 中的那些行，其中“A”和“B”列中的条目出现在 df2 中同名的列中。也就是说，使用 df2 的列“A”和“B”过滤 df1。

通常我认为这将是一个连接（通过合并实现），但是如何将熊猫数据帧与 pyspark 数据帧连接起来？

我负担不起将df1转换为熊猫数据帧的费用。

Answer 1

您可以使用以下代码片段来获得帮助:

df1 = spark.read.csv("/user/me/data1/")
df2 = pd.read_csv("data2.csv", keep_default_na=False)
df3 = df = sqlContext.createDataFrame(df2, schema)
df = df1.join(df3, ["A", "B"])

Answer 2

您可以在将panda数据帧转换为pyspark数据帧时传递模式，如下所示：

from pyspark.sql.types import *
schema = StructType([
StructField("name", StringType(), True),
StructField("age", IntegerType(), True)])
df = sqlContext.createDataFrame(pandas_dataframe, schema)

或者你可以使用我在这个函数中使用的黑客：

def create_spark_dataframe(file_name):
    """
    will return the spark dataframe input pandas dataframe
    """
    pandas_data_frame = pd.read_csv(file_name)
    for col in pandas_data_frame.columns:
      if ((pandas_data_frame[col].dtypes != np.int64) & (pandas_data_frame[col].dtypes != np.float64)):
        pandas_data_frame[col] = pandas_data_frame[col].fillna('')

    spark_data_frame = sqlContext.createDataFrame(pandas_data_frame)
    return spark_data_frame