pyspark.sql中如何创建DataFrame

在pyspark.sql中创建DataFrame有多种方法，可以从不同的数据源创建DataFrame，也可以通过转换已有的RDD来创建DataFrame。

一、从数据源创建DataFrame

1. 从csv文件创建DataFrame

可以使用spark.read.csv()方法从csv文件中创建DataFrame。可以指定文件路径、分隔符、文件格式等信息。

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("Create DataFrame").getOrCreate()

# 从csv文件创建DataFrame
df = spark.read.csv("path/to/file.csv", header=True, inferSchema=True)
df.show()

2. 从数据库表创建DataFrame

可以使用spark.read.jdbc()方法从关系型数据库中创建DataFrame。需要指定数据库URL、表名、用户名、密码等信息。

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("Create DataFrame").getOrCreate()

# 从数据库表创建DataFrame
df = spark.read.jdbc(url="jdbc:mysql://localhost/db_name", table="table_name",
                     properties={"user": "username", "password": "password"})
df.show()

3. 从JSON文件创建DataFrame

可以使用spark.read.json()方法从JSON文件中创建DataFrame。

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("Create DataFrame").getOrCreate()

# 从JSON文件创建DataFrame
df = spark.read.json("path/to/file.json")
df.show()

4. 从Parquet文件创建DataFrame

可以使用spark.read.parquet()方法从Parquet文件中创建DataFrame。

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("Create DataFrame").getOrCreate()

# 从Parquet文件创建DataFrame
df = spark.read.parquet("path/to/file.parquet")
df.show()

5. 从其他数据源创建DataFrame

除了csv、JSON和Parquet文件，还可以从其他数据源如Avro、ORC、Hive表等创建DataFrame，具体方法可参考pyspark.sql的官方文档。

二、通过转换已有的RDD创建DataFrame

如果已经有一个RDD对象，可以通过将其转换为DataFrame来创建DataFrame，需要定义列名和数据类型。

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

spark = SparkSession.builder.appName("Create DataFrame").getOrCreate()

# 创建一个RDD
rdd = spark.sparkContext.parallelize([(1, "Alice"), (2, "Bob"), (3, "Charlie")])

# 定义列名和数据类型
schema = StructType([
    StructField("id", IntegerType(), True),
    StructField("name", StringType(), True)
])

# 将RDD转换为DataFrame
df = spark.createDataFrame(rdd, schema)
df.show()

本篇文章介绍了在pyspark.sql中创建DataFrame的多种方法，包括从不同的数据源创建DataFrame和通过转换已有的RDD创建DataFrame。可以根据具体需求选择合适的方法来创建DataFrame。