spark rdd转dataframe 写入mysql的实例讲解

要将Spark RDD转换成DataFrame，并将其写入MySQL，您可以按照以下步骤进行操作：

第1步：导入库

假设您已经在Spark和MySQL上安装了适当的依赖项。在这个例子中，我们将使用Spark Core，Spark SQL和MySQL connector。请确保将这些库导入到您的代码库中。

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

第2步：创建SparkSession

使用SparkSession对象，您可以从Spark RDD创建DataFrame、执行SQL查询以及将DataFrame写入各种数据源。此外，它使得访问SparkAPI更加简单。

spark = SparkSession.builder.appName('<app_name>').getOrCreate()

第3步：创建RDD

在此步骤中，我们将创建一个简单的RDD，以便将其转换为DataFrame并将其写入到MySQL中去。

rdd = spark.sparkContext.parallelize([(1,'John'),(2,'Mike'),(3,'David')])

第4步：创建DataFrame模式

创建Schema以定义DataFrame中的列，然后将RDD转换为DataFrame。

schema = StructType([StructField("id", IntegerType(),True),StructField("name", StringType(), True)])
df = spark.createDataFrame(rdd,schema)

第5步：将DataFrame写入到MySQL

将DataFrame写入MySQL之前，我们需要按照以下步骤进行配置：

1.设置JDBC驱动程序

2.定义JDBC URL

3.创建连接，使用者的必要连接信息

4.写入DataFrame

以下示例中显示了如何将DataFrame写入MySQL。

properties = {'user': 'root', 'password': '<password>',
              'driver': 'com.mysql.jdbc.Driver'}
df.write.jdbc("jdbc:mysql://localhost:3306/<database_name>", "<table_name>", properties=properties)

第6步：完整代码示例：

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

spark = SparkSession.builder.appName('<app_name>').getOrCreate()

# create rdd
rdd = spark.sparkContext.parallelize([(1,'John'),(2,'Mike'),(3,'David')])

# Define schema for rdd to create DataFrame
schema = StructType([StructField("id", IntegerType(),True),StructField("name", StringType(), True)])
df = spark.createDataFrame(rdd,schema)

properties = {'user': 'root', 'password': '<password>', 'driver': 'com.mysql.jdbc.Driver'}
df.write.jdbc("jdbc:mysql://localhost:3306/<database_name>", "<table_name>", properties=properties)

第7步：示例2

以下示例建立在第一个示例的基础上，并演示了如何从MySQL中读取数据，并将其转换为Spark DataFrame。

# Read data from MySQL and convert into Spark DataFrame
jdbcDF = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:mysql://localhost:3306/<database_name>") \
    .option("driver", "com.mysql.jdbc.Driver") \
    .option("dbtable", "<table_name>") \
    .option("user", "root") \
    .option("password", "<password>") \
    .load()

# Show DataFrame
jdbcDF.show()

完整的代码示例如下：

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

spark = SparkSession.builder.appName('<app_name>').getOrCreate()

# create rdd
rdd = spark.sparkContext.parallelize([(1,'John'),(2,'Mike'),(3,'David')])

# Define schema for rdd to create DataFrame
schema = StructType([StructField("id", IntegerType(),True),StructField("name", StringType(), True)])
df = spark.createDataFrame(rdd,schema)

properties = {'user': 'root', 'password': '<password>', 'driver': 'com.mysql.jdbc.Driver'}
df.write.jdbc("jdbc:mysql://localhost:3306/<database_name>", "<table_name>", properties=properties)

# Read data from MySQL and convert into Spark DataFrame
jdbcDF = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:mysql://localhost:3306/<database_name>") \
    .option("driver", "com.mysql.jdbc.Driver") \
    .option("dbtable", "<table_name>") \
    .option("user", "root") \
    .option("password", "<password>") \
    .load()

# Show DataFrame
jdbcDF.show()