将spark dataframe数据类型映射到jdbc数据类型

from pyspark.sql import SparkSession
from pyspark.sql.types import IntegerType, StringType, DateType

# 创建SparkSession
spark = SparkSession.builder.appName("example").getOrCreate()

# 示例DataFrame
data = [
    (1, "Alice", "2023-01-01"),
    (2, "Bob", "2023-02-01"),
    (3, "Charlie", "2023-03-01")
]
columns = ["id", "name", "dob"]
df = spark.createDataFrame(data, columns)

# 定义JDBC连接参数
jdbc_url = "jdbc:mysql://localhost:3306/mydatabase"
connection_properties = {
    "user": "username",
    "password": "password",
    "driver": "com.mysql.jdbc.Driver"
}

# 将DataFrame写入JDBC数据库
def insert_into_db(partition):
    import pymysql
    conn = None
    try:
        conn = pymysql.connect(**connection_properties, host=jdbc_url.split("//")[1].split(":")[0], port=int(jdbc_url.split(":")[2].split("/")[0]))
        cursor = conn.cursor()
        for row in partition:
            cursor.execute("INSERT INTO mytable (id, name, dob) VALUES (%s, %s, %s)", row)
        conn.commit()
    except Exception as e:
        print(f"Error: {e}")
    finally:
        if conn:
            conn.close()

df.foreachPartition(insert_into_db)

df.write.jdbc(url=jdbc_url, table="mytable", mode="append", properties=connection_properties)