银行每日CSV账户数据增量加载至Redshift的实现方法问询

嘿，这个场景我太熟了——银行账户数据的增量同步是典型的变更数据捕获（CDC）需求，结合你提到的CSV→HDFS→Hive/Spark→Redshift流程，我给你拆解成可落地的步骤，每一步都有实际操作的细节：

首先得确保你的CSV文件有两个核心要素：

• 唯一标识：比如account_id，这是匹配新旧账户数据的关键，没有这个一切免谈。
• 批次时间戳：比如file_generated_ts，用来标记文件的生成日期，方便区分每日的数据集。

然后在HDFS上按日期归档文件，比如存成/user/bank/csv/2024-05-20/account_details.csv，旧文件至少保留1-2周，用来做每日的对比校验。

Hive也能做对比，但Spark的处理速度和灵活性更适合大量账户数据的场景：

• 第一步：加载新旧两日的数据集
  用Spark读取HDFS上的今日和昨日CSV，转换成DataFrame（建议手动指定Schema，避免自动推断的类型错误）：

// 先定义固定Schema，比如：
val accountSchema = StructType(Seq(
  StructField("account_id", StringType, nullable = false),
  StructField("account_name", StringType, nullable = true),
  StructField("balance", DecimalType(18,2), nullable = true),
  StructField("file_generated_ts", TimestampType, nullable = false)
))

val todayDF = spark.read
  .option("header", "true")
  .schema(accountSchema)
  .csv("/user/bank/csv/2024-05-20/*.csv")
  .dropDuplicates("account_id") // 先去重当日重复的账户

val yesterdayDF = spark.read
  .option("header", "true")
  .schema(accountSchema)
  .csv("/user/bank/csv/2024-05-19/*.csv")
  .dropDuplicates("account_id")

注意：如果是首次加载，直接全量导入目标表即可，后续再走增量流程。

• 第二步：识别三种变更类型
  通过join操作区分新增、更新、删除三类数据：

  • 新增账户：今日存在但昨日没有的账户ID

  val newAccountsDF = todayDF.join(yesterdayDF, Seq("account_id"), "leftanti")
  

  • 更新账户：账户ID两日都存在，但至少一个业务字段有变化（排除file_generated_ts这类批次字段）

  // 筛选出需要对比的业务字段
  val compareCols = todayDF.columns.filter(!Set("account_id", "file_generated_ts").contains(_))
  // 对比字段是否有差异
  val updateCondition = compareCols.map(col => todayDF(col) =!= yesterdayDF(col)).reduce(_ || _)
  val updatedAccountsDF = todayDF.join(yesterdayDF, Seq("account_id"), "inner")
    .where(updateCondition)
    .select(todayDF("*")) // 取今日的最新数据
  

  • 删除账户：昨日存在但今日没有的账户ID（这里要和业务确认规则：是真删除还是文件漏发？比如可以要求连续2日未出现才标记删除，或者CSV里有明确的delete_flag）

  val deletedAccountsDF = yesterdayDF.join(todayDF, Seq("account_id"), "leftanti")
    .select("account_id") // 只需要账户ID用来在Redshift处理
  

• 第三步：把变更数据写入临时层
  把新增和更新的数据合并，删除数据单独存储，写入HDFS的Parquet文件（比CSV更适合后续加载）：

val combinedChangesDF = newAccountsDF.union(updatedAccountsDF)
// 写入临时目录，供Redshift加载
combinedChangesDF.write.mode("overwrite").parquet("/user/bank/staging/changed_accounts/")
deletedAccountsDF.write.mode("overwrite").parquet("/user/bank/staging/deleted_accounts/")

Redshift本身支持高效的批量加载，结合变更数据可以选下面两种方式：

方案A：用COPY命令+MERGE（推荐，大数量场景） #

这是Redshift官方推荐的批量加载方式，效率最高：

• 第一步：加载新增/更新数据到临时表
  先把HDFS上的变更数据同步到S3（可以用DistCp），然后用COPY命令加载到Redshift临时表：

-- 创建和目标表结构一致的临时表
CREATE TEMP TABLE tmp_changed_accounts (LIKE target_accounts);

-- 从S3加载Parquet数据
COPY tmp_changed_accounts 
FROM 's3://your-bucket/path/to/changed_accounts/'
IAM_ROLE 'arn:aws:iam::123456789012:role/RedshiftS3Access'
FORMAT AS PARQUET;

-- 用MERGE命令同步到目标表（Redshift 1.0.1953及以上版本支持）
MERGE INTO target_accounts t
USING tmp_changed_accounts s
ON t.account_id = s.account_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

• 第二步：处理删除数据
  如果业务要求软删除（标记而非物理删除）：

-- 先加载删除的账户ID到临时表
CREATE TEMP TABLE tmp_deleted_accounts (account_id VARCHAR);
COPY tmp_deleted_accounts 
FROM 's3://your-bucket/path/to/deleted_accounts/'
IAM_ROLE 'arn:aws:iam::123456789012:role/RedshiftS3Access'
FORMAT AS PARQUET;

-- 更新删除标记
UPDATE target_accounts
SET is_deleted = TRUE, updated_at = CURRENT_TIMESTAMP
WHERE account_id IN (SELECT account_id FROM tmp_deleted_accounts);

如果是硬删除（谨慎使用，不可逆）：

DELETE FROM target_accounts
WHERE account_id IN (SELECT account_id FROM tmp_deleted_accounts);

方案B：用Spark直接写入Redshift（小批量场景） #

如果每日变更数据量不大，可以用Spark的Redshift连接器直接写入：

combinedChangesDF.write
  .format("com.databricks.spark.redshift")
  .option("url", "jdbc:redshift://your-redshift-cluster:5439/your-db?user=xxx&password=xxx")
  .option("dbtable", "target_accounts")
  .option("tempdir", "s3://your-bucket/temp/") // Spark需要临时目录来上传数据
  .mode("append") // 新增用append，更新的话可以先删对应ID再插入
  .save()

注意：这种方式对于大数据量，效率不如COPY命令，适合小批量变更。

• Schema一致性检查：每日CSV的字段必须和预定义的Schema一致，否则会导致读取失败。可以在加载前先做Schema校验，比如用Spark的schema方法强制匹配。
• 重复数据处理：每日CSV可能存在重复的account_id，一定要先做dropDuplicates("account_id")，避免后续同步出错。
• 增量校验：每次同步后，统计变更数据的数量（新增多少、更新多少、删除多少），和业务预期对比，确保数据准确。
• 历史数据归档：HDFS上的旧CSV文件可以定期归档到冷存储（比如S3 Glacier），节省存储成本。

内容的提问来源于stack exchange，提问作者Ram