大数据开发！Pandas转spark无痛指南！⛵

# pandas vs pyspark，工具库导入
import pandas as pd
import pyspark.sql.functions as F

from pyspark.sql import SparkSession
spark = SparkSession\
.builder\
.appName('SparkByExamples.com')\
.getOrCreate()

columns = ["employee","department","state","salary","age"]
data = [("Alain","Sales","Paris",60000,34),
        ("Ahmed","Sales","Lyon",80000,45),
        ("Ines","Sales","Nice",55000,30),
        ("Fatima","Finance","Paris",90000,28),
        ("Marie","Finance","Nantes",100000,40)]

df = pd.DataFrame(data=data, columns=columns)
# 查看头2行
df.head(2)

df = spark.createDataFrame(data).toDF(*columns)
# 查看头2行
df.limit(2).show()

types_dict = {
    "employee": pd.Series([r[0] for r in data], dtype='str'),
    "department": pd.Series([r[1] for r in data], dtype='str'),
    "state": pd.Series([r[2] for r in data], dtype='str'),
    "salary": pd.Series([r[3] for r in data], dtype='int'),
    "age": pd.Series([r[4] for r in data], dtype='int')
}

df = pd.DataFrame(types_dict)

df.dtypes

from pyspark.sql.types import StructType,StructField, StringType, IntegerType

schema = StructType([ \
    StructField("employee",StringType(),True), \
    StructField("department",StringType(),True), \
    StructField("state",StringType(),True), \
    StructField("salary", IntegerType(), True), \
    StructField("age", IntegerType(), True) \
  ])

df = spark.createDataFrame(data=data,schema=schema)

df.dtypes
# 查看数据类型 
df.printSchema() 

df = pd.read_csv(path, sep=';', header=True)
df.to_csv(path, ';', index=False)

df = spark.read.csv(path, sep=';')
df.coalesce(n).write.mode('overwrite').csv(path, sep=';')

df.partitionBy("department","state").write.mode('overwrite').csv(path, sep=';')

columns_subset = ['employee', 'salary']

df[columns_subset].head()

df.loc[:, columns_subset].head()

columns_subset = ['employee', 'salary']

df.select(columns_subset).show(5)

# 头2行
df.iloc[:2].head()

df.take(2).head()
# 或者
df.limit(2).head()

# First method
flt = (df['salary'] >= 90_000) & (df['state'] == 'Paris')
filtered_df = df[flt]

# Second Method: Using query which is generally faster
filtered_df = df.query('(salary >= 90_000) and (state == "Paris")')
# Or
target_state = "Paris"
filtered_df = df.query('(salary >= 90_000) and (state == @target_state)')

# 方法1：基于filter进行数据选择
filtered_df = df.filter((F.col('salary') >= 90_000) & (F.col('state') == 'Paris'))

# 或者
filtered_df = df.filter(F.expr('(salary >= 90000) and (state == "Paris")'))

# 方法2：基于SQL进行数据选择
df.createOrReplaceTempView("people")

filtered_df = spark.sql("""
SELECT * FROM people
WHERE (salary >= 90000) and (state == "Paris")
""") 

seniority = [3, 5, 2, 4, 10]
# 方法1
df['seniority'] = seniority

# 方法2
df.insert(2, "seniority", seniority, True)

seniority = [3, 5, 2, 4, 10]
df = df.withColumn('seniority', seniority)

# pandas拼接2个dataframe
df_to_add = pd.DataFrame(data=[("Robert","Advertisement","Paris",55000,27)], columns=columns)
df = pd.concat([df, df_to_add], ignore_index = True)

# PySpark拼接2个dataframe
df_to_add = spark.createDataFrame([("Robert","Advertisement","Paris",55000,27)]).toDF(*columns)
df = df.union(df_to_add)

# pandas拼接多个dataframe
dfs = [df, df1, df2,...,dfn]
df = pd.concat(dfs, ignore_index = True)

# pyspark拼接多个dataframe
from functools import reduce
from pyspark.sql import DataFrame

def unionAll(*dfs):
    return reduce(DataFrame.unionAll, dfs)

dfs = [df, df1, df2,...,dfn]
df = unionAll(*dfs)

df.summary()
#或者
df.describe()

df.groupby('department').agg({'employee': 'count', 'salary':'max', 'age':'mean'})

df.groupBy('department').agg({'employee': 'count', 'salary':'max', 'age':'mean'})

df.groupby('department').agg({'employee': 'count', 'salary':'max', 'age':'mean'}).reset_index()

df.groupBy('department').agg(F.count('employee').alias('employee'), F.max('salary').alias('salary'), F.mean('age').alias('age'))

df['new_salary'] = df['salary'].apply(lambda x: x*1.15 if x<= 60000 else x*1.05)

from pyspark.sql.types import FloatType

df.withColumn('new_salary', F.udf(lambda x: x*1.15 if x<= 60000 else x*1.05, FloatType())('salary'))