在python的spark环境中将DataFrame数据插入PostgreSQL数据库

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

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

# 定义用户信息列表
users = [
    {'id': 1, 'name': '张三', 'gender': '男', 'birthday': '19900101'},
    {'id': 2, 'name': '李四', 'gender': '女', 'birthday': '19921231'},
    {'id': 3, 'name': '王五', 'gender': '男', 'birthday': '19870615'}
]

# 定义dataframe的schema
schema = StructType([
    StructField("id", IntegerType(), True),
    StructField("name", StringType(), True),
    StructField("gender", StringType(), True),
    StructField("birthday_str", StringType(), True)
])

# 将用户信息列表转换为dataframe
df = spark.createDataFrame(users, schema=schema)

# 将birthday_str列转换为日期类型
df = df.withColumn('birthday', to_date(df['birthday_str'], 'yyyyMMdd'))

# 删除birthday_str列
df = df.drop('birthday_str')

# 打印dataframe
df.show()

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

# 创建SparkSession
spark = SparkSession.builder.appName("Insert Data to PostgreSQL").getOrCreate()

# 定义用户信息列表
users = [
    {'id': 1, 'name': '张三', 'gender': '男', 'birthday': '1990-01-01'},
    {'id': 2, 'name': '李四', 'gender': '女', 'birthday': '1995-03-15'},
    {'id': 3, 'name': '王五', 'gender': '男', 'birthday': '1985-12-21'},
    # 其他17条数据
]

# 定义dataframe的schema
schema = StructType([
    StructField("id", IntegerType(), True),
    StructField("name", StringType(), True),
    StructField("gender", StringType(), True),
    StructField("birthday", StringType(), True)
])

# 将用户信息列表转换为dataframe
df = spark.createDataFrame(users, schema=schema)

# 将dataframe中的数据插入到PostgreSQL数据库中，并在插入过程中处理异常
def write_to_postgresql(rows):
    # 连接数据库
    conn = psycopg2.connect(
        host='localhost',
        port='5432',
        dbname='test_db',
        user='test_user',
        password='test_password'
    )
    cur = conn.cursor()
    try:
        # 插入数据
        for row in rows:
            try:
                cur.execute(
                    "INSERT INTO users(id, name, gender, birthday) VALUES (%s, %s, %s, %s)",
                    (row.id, row.name, row.gender, row.birthday)
                )
            except:
                print(f"Error inserting row {row.id}")
                raise
            # 每插入10条数据提交一次事务
            if cur.rowcount % 10 == 0:
                conn.commit()
        # 提交剩余的数据
        conn.commit()
    except:
        conn.rollback()
        raise
    finally:
        # 关闭数据库连接
        cur.close()
        conn.close()

# 将dataframe中的数据按照每个分区分别插入到数据库中
df.foreachPartition(write_to_postgresql)

# 将DataFrame写入PostgreSQL数据库中
dataframe.write \
    .mode('overwrite') \  # 覆盖模式，如果表已经存在，则删除并重新创建
    .format("jdbc") \  # 指定使用JDBC连接器
    .option("url", "jdbc:postgresql://<HOST>:<PORT>/<DATABASE>") \  # 指定PostgreSQL数据库的URL
    .option("dbtable", "<TABLE_NAME>") \  # 指定表名
    .option("user", "<USERNAME>") \  # 指定数据库用户名
    .option("password", "<PASSWORD>") \  # 指定数据库密码
    .option("driver", "org.postgresql.Driver") \  # 指定JDBC驱动程序
    .option("batchsize", "10000") \  # 指定每批次写入的数据量
    .option("truncate", "true") \  # 如果表已经存在，则截断表
    .option("isolationLevel", "NONE") \  # 事务隔离级别，这里设置为NONE
    .option("rewriteBatchedStatements", "true") \  # 开启批量处理
    .save()

# 日文翻译：
# PostgreSQLデータベースにDataFrameを書き込む
dataframe.write
  .mode('overwrite') \ # オーバーライトモード。テーブルが既に存在する場合は削除して再作成する
  .format("jdbc") \ # JDBCコネクタを使用することを指定
  .option("url", "jdbc:postgresql://<HOST>:<PORT>/<DATABASE>") \ # PostgreSQLデータベースのURLを指定
  .option("dbtable", "<TABLE_NAME>") \ # テーブル名を指定
  .option("user", "<USERNAME>") \ # データベースのユーザー名を指定
  .option("password", "<PASSWORD>") \ # データベースのパスワードを指定
  .option("driver", "org.postgresql.Driver") \ # JDBCドライバを指定
  .option("batchsize", "10000") \ # 1バッチあたりの書き込みデータ量を指定
  .option("truncate", "true") \ # テーブルが既に存在する場合は切り捨てる
  .option("isolationLevel", "NONE") \ # トランザクションの分離レベル。ここではNONEに設定する
  .option("rewriteBatchedStatements", "true") \ # バッチ処理を有効にする
  .save()