Python大数据分析之PySpark原理与实战教程详解

pip install pyspark

python -c "import pyspark; print(pyspark.__version__)"
pyspark

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("PySparkDemo").getOrCreate()

# 读取CSV文件
df = spark.read.csv("data.csv", header=True, inferSchema=True)
# 保存为Parquet格式
df.write.parquet("output.parquet")

from pyspark.sql.functions import col
# 选择、过滤、添加新列
df2 = df.select("name", "age").filter(col("age") > 18)
df2 = df2.withColumn("age_group", (col("age")/10).cast("int")*10)

df.groupBy("age_group").count().show()

df.createOrReplaceTempView("people")
spark.sql("SELECT age_group, COUNT(*) FROM people GROUP BY age_group").show()

pandas_df = df.toPandas()
import matplotlib.pyplot as plt
pandas_df['age'].hist()
plt.show()

from pyspark.ml.feature import VectorAssembler
from pyspark.ml.classification import LogisticRegression

# 特征组装
assembler = VectorAssembler(inputCols=["age", "income"], outputCol="features")
train_df = assembler.transform(df)

# 逻辑回归模型
lr = LogisticRegression(featuresCol="features", labelCol="label")
model = lr.fit(train_df)
result = model.transform(train_df)
result.select("prediction", "label").show()

from pyspark.sql.types import StructType, StringType, IntegerType
schema = StructType().add("name", StringType()).add("age", IntegerType())
stream_df = spark.readStream.schema(schema).csv("input_dir/")
query = stream_df.writeStream.format("console").start()
query.awaitTermination()