我正在尝试将Pandas DF转换为Spark one。DF头：
10000001,1,0,1,12:35,OK,10002,1,0,9,f,NA,24,24,0,3,9,0,0,1,1,0,0,4,543
10000001,2,0,1,12:36,OK,10002,1,0,9,f,NA,24,24,0,3,9,2,1,1,3,1,3,2,611
10000002,1,0,4,12:19,PA,10003,1,1,7,f,NA,74,74,0,2,15,2,0,2,3,1,2,2,691
法典：
dataset = pd.read_csv("data/AS/test_v2.csv")
sc = SparkContext(conf=conf)
sqlCtx = SQLContext(sc)
sdf = sqlCtx.createDataFrame(dataset)
我收到一个错误：
TypeError: Can not merge type <class 'pyspark.sql.types.StringType'> and <class 'pyspark.sql.types.DoubleType'>
我制作了这个脚本，它适用于我的 10 个熊猫数据帧
from pyspark.sql.types import *
# Auxiliar functions
def equivalent_type(f):
if f == 'datetime64[ns]': return TimestampType()
elif f == 'int64': return LongType()
elif f == 'int32': return IntegerType()
elif f == 'float64': return DoubleType()
elif f == 'float32': return FloatType()
else: return StringType()
def define_structure(string, format_type):
try: typo = equivalent_type(format_type)
except: typo = StringType()
return StructField(string, typo)
# Given pandas dataframe, it will return a spark's dataframe.
def pandas_to_spark(pandas_df):
columns = list(pandas_df.columns)
types = list(pandas_df.dtypes)
struct_list = []
for column, typo in zip(columns, types):
struct_list.append(define_structure(column, typo))
p_schema = StructType(struct_list)
return sqlContext.createDataFrame(pandas_df, p_schema)
你也可以在这个要点中看到它
有了这个，你只需要打电话spark_df = pandas_to_spark(pandas_df)
您需要确保 pandas 数据帧列适合 Spark 推断的类型。 如果熊猫数据帧列出类似以下内容：
pd.info()
<class 'pandas.core.frame.DataFrame'>
RangeIndex: 5062 entries, 0 to 5061
Data columns (total 51 columns):
SomeCol 5062 non-null object
Col2 5062 non-null object
你得到这个错误，尝试：
df[['SomeCol', 'Col2']] = df[['SomeCol', 'Col2']].astype(str)
现在，确保.astype(str)实际上是您希望这些列的类型。 基本上，当底层 Java 代码试图从 python 中的对象推断类型时，它会使用一些观察结果并进行猜测，如果该猜测不适用于列中的所有数据，它试图从熊猫转换为 spark，它将失败。
可以通过按如下方式强加架构来避免与类型相关的错误：
注意：使用原始数据（如上所述）创建了一个文本文件（test.csv），并插入了假设的列名（"Col1"，"Col2",...,"Col25"）。
import pyspark
from pyspark.sql import SparkSession
import pandas as pd
spark = SparkSession.builder.appName('pandasToSparkDF').getOrCreate()
pdDF = pd.read_csv("test.csv")
熊猫数据框的内容：
col1 col2 col3 col4 col5 col6 col7 col8 ...
0 10000001 1 0 1 12:35 OK 10002 1 ...
1 10000001 2 0 1 12:36 OK 10002 1 ...
2 10000002 1 0 4 12:19 PA 10003 1 ...
接下来，创建架构：
from pyspark.sql.types import *
mySchema = StructType([ StructField("col1", LongType(), True)
,StructField("col2", IntegerType(), True)
,StructField("col3", IntegerType(), True)
,StructField("col4", IntegerType(), True)
,StructField("col5", StringType(), True)
,StructField("col6", StringType(), True)
,StructField("col7", IntegerType(), True)
,StructField("col8", IntegerType(), True)
,StructField("col9", IntegerType(), True)
,StructField("col10", IntegerType(), True)
,StructField("col11", StringType(), True)
,StructField("col12", StringType(), True)
,StructField("col13", IntegerType(), True)
,StructField("col14", IntegerType(), True)
,StructField("col15", IntegerType(), True)
,StructField("col16", IntegerType(), True)
,StructField("col17", IntegerType(), True)
,StructField("col18", IntegerType(), True)
,StructField("col19", IntegerType(), True)
,StructField("col20", IntegerType(), True)
,StructField("col21", IntegerType(), True)
,StructField("col22", IntegerType(), True)
,StructField("col23", IntegerType(), True)
,StructField("col24", IntegerType(), True)
,StructField("col25", IntegerType(), True)])
注意：True（表示允许可为空）
创建 PySpark 数据帧：
df = spark.createDataFrame(pdDF,schema=mySchema)
确认 Pandas 数据框现在是 PySpark 数据框：
type(df)
输出：
pyspark.sql.dataframe.DataFrame
旁白：
要解决 Kate 在下面的评论 - 要强加通用（字符串）架构，您可以执行以下操作：
df=spark.createDataFrame(pdDF.astype(str))
在 Spark 版本>= 3 中，您可以在一行中将 pandas 数据帧转换为 pyspark 数据帧
use spark.createDataFrame（pandasDF）
dataset = pd.read_csv("data/AS/test_v2.csv")
sparkDf = spark.createDataFrame(dataset);
如果您对 Spark 会话变量感到困惑，火花会话如下
sc = SparkContext.getOrCreate(SparkConf().setMaster("local[*]"))
spark = SparkSession
.builder
.getOrCreate()
我已经用您的数据尝试过这个，它正在工作：
%pyspark
import pandas as pd
from pyspark.sql import SQLContext
print sc
df = pd.read_csv("test.csv")
print type(df)
print df
sqlCtx = SQLContext(sc)
sqlCtx.createDataFrame(df).show()
我清理/简化了顶部答案：
import pyspark.sql.types as ps_types
def get_equivalent_spark_type(pandas_type):
"""
This method will retrieve the corresponding spark type given a pandas
type.
Args:
pandas_type (str): pandas data type
Returns:
spark data type
"""
type_map = {
'datetime64[ns]': ps_types.TimestampType(),
'int64': ps_types.LongType(),
'int32': ps_types.IntegerType(),
'float64': ps_types.DoubleType(),
'float32': ps_types.FloatType()}
if pandas_type not in type_map:
return ps_types.StringType()
else:
return type_map[pandas_type]
def pandas_to_spark(spark, pandas_df):
"""
This method will return a spark dataframe given a pandas dataframe.
Args:
spark (pyspark.sql.session.SparkSession): pyspark session
pandas_df (pandas.core.frame.DataFrame): pandas DataFrame
Returns:
equivalent spark DataFrame
"""
columns = list(pandas_df.columns)
types = list(pandas_df.dtypes)
p_schema = ps_types.StructType([
ps_types.StructField(column, get_equivalent_spark_type(pandas_type))
for column, pandas_type in zip(columns, types)])
return spark.createDataFrame(pandas_df, p_schema)