如何使用Java Spark避免向HDFS写入重复Parquet数据

针对你遇到的重复日志写入问题，以下是几个实用的落地方案，覆盖行级、文件级以及实时流场景的去重需求：

1. 基于事件唯一标识的行级去重 #

这是最通用的方案，核心是利用日志中的唯一标识（比如事件ID、请求UUID、日志生成的唯一ID），对比已有数据过滤重复项。

适用场景 #

• 日志中包含全局唯一的事件标识；
• 每次处理的日志中既有重复数据又有新数据。

实现代码（Java Spark） #

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;

public class RowLevelDeduplication {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
                .appName("LogRowDeduplication")
                .getOrCreate();

        // 1. 读取并解析新日志（parse_log为自定义UDF，将文本转为结构化数据）
        Dataset<Row> newLogs = spark.read().text("hdfs://your-log-path")
                .selectExpr("parse_log(value) as event")
                .select("event.event_id", "event.timestamp", "event.content");

        // 2. 读取已有的Parquet历史数据
        Dataset<Row> existingData = spark.read().parquet("hdfs://your-parquet-path");

        // 3. 左反连接：仅保留新日志中未出现在历史数据的记录
        Dataset<Row> uniqueNewData = newLogs.join(
                existingData,
                newLogs.col("event_id").equalTo(existingData.col("event_id")),
                "left_anti"
        );

        // 4. 追加写入HDFS
        uniqueNewData.write().mode(SaveMode.Append).parquet("hdfs://your-parquet-path");

        spark.stop();
    }
}

注意事项 #

• 如果日志没有自带唯一标识，可通过多字段组合+哈希生成：

  newLogs = newLogs.withColumn("unique_key", sha2(concat(col("timestamp"), col("content")), 256));
  

• 左反连接比dropDuplicates更适合增量场景，能直接对比历史数据，避免漏判跨批次的重复项。

2. 基于文件标识的文件级去重 #

如果日志是按文件批量发送（比如每10小时发一个文件），可以通过记录已处理的文件标识来避免重复处理整个文件。

适用场景 #

• 重复发送整个日志文件；
• 日志文件滚动但可通过唯一标识追踪（如HDFS文件ID、etag）。

实现代码（Java Spark + HDFS API） #

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;

import java.io.*;
import java.util.*;

public class FileLevelDeduplication {
    public static void main(String[] args) throws IOException {
        SparkSession spark = SparkSession.builder()
                .appName("LogFileDeduplication")
                .getOrCreate();

        FileSystem fs = FileSystem.get(spark.sparkContext().hadoopConfiguration());
        Path logDir = new Path("hdfs://your-log-dir");
        Path processedRecordPath = new Path("hdfs://your-meta-path/processed_files.txt");

        // 1. 加载已处理的文件标识列表
        Set<String> processedFileIds = new HashSet<>();
        if (fs.exists(processedRecordPath)) {
            BufferedReader reader = new BufferedReader(new InputStreamReader(fs.open(processedRecordPath)));
            String line;
            while ((line = reader.readLine()) != null) {
                processedFileIds.add(line);
            }
            reader.close();
        }

        // 2. 筛选未处理的日志文件
        FileStatus[] allLogFiles = fs.listStatus(logDir);
        List<String> unprocessedPaths = new ArrayList<>();
        for (FileStatus file : allLogFiles) {
            String fileId = String.valueOf(file.getFileId()); // 用HDFS唯一文件ID作为标识
            if (!processedFileIds.contains(fileId)) {
                unprocessedPaths.add(file.getPath().toString());
                processedFileIds.add(fileId);
            }
        }

        // 3. 处理未处理的文件
        if (!unprocessedPaths.isEmpty()) {
            Dataset<Row> newLogs = spark.read().text(unprocessedPaths.toArray(new String[0]))
                    .selectExpr("parse_log(value) as event")
                    .select("event.event_id", "event.timestamp", "event.content");

            newLogs.write().mode(SaveMode.Append).parquet("hdfs://your-parquet-path");

            // 4. 更新已处理文件记录
            BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(fs.create(processedRecordPath, true)));
            for (String id : processedFileIds) {
                writer.write(id);
                writer.newLine();
            }
            writer.close();
        }

        spark.stop();
    }
}

注意事项 #

• 优先用HDFS的fileId或etag作为文件标识，比路径更稳定（避免文件重命名导致的误判）；
• 大文件不建议计算MD5，会带来较高性能开销。

3. 用Spark结构化流实现Exactly-Once语义 #

如果日志是持续产生的（比如实时滚动日志），用Spark Structured Streaming的Checkpoint机制可以保证精确一次处理，从根源避免重复写入。

适用场景 #

• 实时或准实时的日志处理；
• 需要长期稳定的增量处理任务。

实现代码（Java Spark） #

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.SparkSession;

public class StreamingDeduplication {
    public static void main(String[] args) throws InterruptedException {
        SparkSession spark = SparkSession.builder()
                .appName("StreamingLogDeduplication")
                .getOrCreate();

        // 1. 监控目录下的新增日志文件
        Dataset<Row> logStream = spark.readStream()
                .format("text")
                .load("hdfs://your-log-dir");

        // 2. 解析日志并按事件ID去重
        Dataset<Row> processedStream = logStream.selectExpr("parse_log(value) as event")
                .select("event.event_id", "event.timestamp", "event.content")
                .dropDuplicates("event_id");

        // 3. 写入Parquet，开启Checkpoint保证Exactly-Once
        StreamingQuery query = processedStream.writeStream()
                .format("parquet")
                .option("path", "hdfs://your-parquet-path")
                .option("checkpointLocation", "hdfs://your-checkpoint-path") // 必须设置，记录处理进度
                .outputMode("append")
                .start();

        query.awaitTermination();
    }
}

注意事项 #

• Checkpoint目录由Spark自动维护，禁止手动删除，否则会丢失处理进度导致重复处理；
• 配合Watermark可处理迟到数据，进一步优化去重精度：

  processedStream = processedStream.withWatermark("timestamp", "1 hour")
          .dropDuplicates("event_id", "timestamp");
  

额外优化：原子写入保证数据一致性 #

写入HDFS时，先写入临时目录，完成后再原子重命名到正式路径，避免任务故障导致的部分写入或重复写入：

// 先写入临时路径
String tempPath = "hdfs://your-parquet-path/temp_" + System.currentTimeMillis();
uniqueNewData.write().mode(SaveMode.Overwrite).parquet(tempPath);

// 原子重命名到正式路径
Path tempDir = new Path(tempPath);
Path targetDir = new Path("hdfs://your-parquet-path");
FileSystem fs = FileSystem.get(spark.sparkContext().hadoopConfiguration());
if (fs.exists(targetDir)) {
    Path newSubDir = new Path(targetDir, "batch_" + System.currentTimeMillis());
    fs.rename(tempDir, newSubDir);
} else {
    fs.rename(tempDir, targetDir);
}

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