python - 如何将数据从 pandas 数据帧分块加载到 spark 数据帧

Question

我已经使用类似这样的方法通过 pyodbc 连接读取数据 block :

import pandas as pd
import pyodbc
conn = pyodbc.connect("Some connection Details")
sql = "SELECT * from TABLES;"
df1 = pd.read_sql(sql,conn,chunksize=10)

现在我想使用类似的方法将所有这些 block 读入一个单一的 spark 数据帧中:

i = 0
for chunk in df1:
    if i==0:
        df2 = sqlContext.createDataFrame(chunk)
    else:
        df2.unionAll(sqlContext.createDataFrame(chunk))
    i = i+1

问题是当我执行 df2.count() 时，我得到的结果是 10，这意味着只有 i=0 的情况有效。这是 unionAll 的错误吗？我在这里做错了什么吗？？

Answer 1

.unionAll() 的文档声明它返回一个新的数据帧，因此您必须分配回 df2 DataFrame:

i = 0
for chunk in df1:
    if i==0:
        df2 = sqlContext.createDataFrame(chunk)
    else:
        df2 = df2.unionAll(sqlContext.createDataFrame(chunk))
    i = i+1

此外，您还可以使用 enumerate()避免必须自己管理 i 变量:

for i,chunk in enumerate(df1):
    if i == 0:
        df2 = sqlContext.createDataFrame(chunk)
    else:
        df2 = df2.unionAll(sqlContext.createDataFrame(chunk))

此外，.unionAll() 的文档指出 .unionAll() 已弃用，现在您应该使用 .union()它的作用类似于 SQL 中的 UNION ALL:

for i,chunk in enumerate(df1):
    if i == 0:
        df2 = sqlContext.createDataFrame(chunk)
    else:
        df2 = df2.union(sqlContext.createDataFrame(chunk))

编辑:
此外，我将停止再说，但在我再说之前:正如@zero323 所说，我们不要在循环中使用 .union() 。让我们改为执行以下操作:

def unionAll(*dfs):
    ' by @zero323 from here: http://stackoverflow.com/a/33744540/42346 '
    first, *rest = dfs  # Python 3.x, for 2.x you'll have to unpack manually
    return first.sql_ctx.createDataFrame(
        first.sql_ctx._sc.union([df.rdd for df in dfs]),
        first.schema
    )

df_list = []
for chunk in df1:
    df_list.append(sqlContext.createDataFrame(chunk))

df_all = unionAll(df_list)