如何在PySpark中添加MySQL详细信息作为属性？

pip install pyspark
pip install mysql-connector-python

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

spark = SparkSession.builder \
    .appName("Add MySQL details as properties in PySpark") \
    .getOrCreate()

mysql_config = {
    "url": "jdbc:mysql://<MySQL_Host>:<MySQL_Port>/<MySQL_Database>",
    "driver": "com.mysql.jdbc.Driver",
    "dbtable": "<MySQL_Table>",
    "user": "<MySQL_Username>",
    "password": "<MySQL_Password>"
}

df = spark.createDataFrame(spark.sparkContext.emptyRDD(), StructType([]))

df = df\
    .withColumn("url", lit(mysql_config["url"]))\
    .withColumn("driver", lit(mysql_config["driver"]))\
    .withColumn("dbtable", lit(mysql_config["dbtable"]))\
    .withColumn("user", lit(mysql_config["user"]))\
    .withColumn("password", lit(mysql_config["password"]))

df.createOrReplaceTempView("my_table")

query = "SELECT * FROM my_table"
result = spark.sql(query)

# 打印结果
result.show()