Java Spark中 mapPartitions的重试在任务失败时导致向BigQuery的重复写入
我有一个在Dataproc上运行的Java Spark作业，它按分区处理数据集并将行分成批次插入。
代码使用rdd.mapPartitions，带迭代器，将行拆分成批次（例如每批100行）。
在迭代器内部，每个批次通过自定义的BigQuery插入方法进行插入（将批量行转换为基于JSON的文件，写入GCP存储桶，并使用BigQuery加载作业配置进行加载）。
有时，Spark任务因存储异常而失败（“Read timed out\tat com.google.cloud.storage.StorageException.translate(StorageException.java:118)”）。任务在以下行失败：
try (WritableByteChannel channel = storage.writer(blobInfo);
当任务失败时，Spark会重新执行整个mapPartitions任务，而不是只重试失败的分区批次（也就是出错的点）。例如，实际正在处理总共3333行，而其中2200行已成功插入，Spark在任务失败时会重跑整个任务（再次处理整个数据集），从而导致重复数据。（2200 + 3333 = 5533）。实际插入的数据记录数应为3333条，多出2200条。
我们的批处理作业每天运行24小时。这个问题只在某些按小时运行的作业中偶发，当出现上述Spark任务失败时才会发生。Dataproc作业本身并不会失败。请告知如何解决该问题，以避免数据插入重复。
来自日志浏览器的错误（警告）：
"message": "Lost task 0.0 in stage 9.0 (TID 13) (
com.google.cloud.storage.spi.v1.HttpStorageRpc.translate(HttpStorageRpc.java:287)\tat com.google.cloud.storage.spi.v1.HttpStorageRpc.open(HttpStorageRpc.java:1007)\tat com.google.cloud.storage.ResumableMedia.lambda$null$0(ResumableMedia.java:37)\tat com.google.api.gax.retrying.DirectRetryingExecutor.submit(DirectRetryingExecutor.java:103)\tat com.google.cloud.RetryHelper.run(RetryHelper.java:76)\tat com.google.cloud.RetryHelper.runWithRetries(RetryHelper.java:50)\tat com.google.cloud.storage.Retrying.run(Retrying.java:54)\tat com.google.cloud.storage.ResumableMedia.lambda$startUploadForBlobInfo$1(ResumableMedia.java:34)\tat com.google.cloud.storage.BlobWriteChannel$Builder.build(BlobWriteChannel.java:316)\tat com.google.cloud.storage.StorageImpl.writer(StorageImpl.java:609)\tat com.google.cloud.storage.StorageImpl.writer(StorageImpl.java:579)\tat com.google.cloud.storage.StorageImpl.writer(StorageImpl.java:101)\tat com.db.lrc.macs.ecomms.ops.preingestion.pipeline.OpsEmailSparkPipeline.insertDataIntoBigQuery(OpsEmailSparkPipeline.java:359)\tat com.db.lrc.macs.ecomms.ops.preingestion.pipeline.OpsEmailSparkPipeline.processBodyContent(OpsEmailSparkPipeline.java:225)\tat com.db.lrc.macs.ecomms.ops.preingestion.pipeline.OpsEmailSparkPipeline.lambda$processInBatches$266d1df4$1(OpsEmailSparkPipeline.java:176)\tat org.apache.spark.api.java.JavaRDDLike.$anonfun$mapPartitions$1(JavaRDDLike.scala:153)\tat org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2(RDD.scala:855)\tat org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2$adapted(RDD.scala:855)\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\tat org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)\tat org.apache.spark.scheduler.Task.run(Task.scala:136)\tat org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:548)\tat org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1504)\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:551)\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\tat java.base/java.lang.Thread.run(Thread.java:829)Caused by: java.net.SocketTimeoutException: Read timed out\tat java.base/java.net.SocketInputStream.socketRead0(Native Method)\tat java.base/java.net.SocketInputStream.socketRead(SocketInputStream.java:115)\tat java.base/java.net.SocketInputStream.read(SocketInputStream.java:168)\tat java.base/java.net.SocketInputStream.read(SocketInputStream.java:140)\tat org.conscrypt.ConscryptEngineSocket$SSLInputStream.readFromSocket(ConscryptEngineSocket.java:920)\tat org.conscrypt.ConscryptEngineSocket$SSLInputStream.processDataFromSocket(ConscryptEngineSocket.java:884)\tat org.conscrypt.ConscryptEngineSocket$SSLInputStream.readUntilDataAvailable(ConscryptEngineSocket.java:799)\tat org.conscrypt.ConscryptEngineSocket$SSLInputStream.read(ConscryptEngineSocket.java:772)\tat java.base/java.io.BufferedInputStream.fill(BufferedInputStream.java:252)\tat java.base/java.io.BufferedInputStream.read1(BufferedInputStream.java:292)\tat java.base/java.io.BufferedInputStream.read(BufferedInputStream.java:351)\tat java.base/sun.net.www.http.HttpClient.parseHTTPHeader(HttpClient.java:789)\tat java.base/sun.net.www.http.HttpClient.parseHTTP(HttpClient.java:724)\tat java.base/sun.net.www.protocol.http.HttpURLConnection.getInputStream0(HttpURLConnection.java:1615)\tat java.base/sun.net.www.protocol.http.HttpURLConnection.getInputStream(HttpURLConnection.java:1520)\tat java.base/java.net.HttpURLConnection.getResponseCode(HttpURLConnection.java:527)\tat java.base/sun.net.www.protocol.https.HttpsURLConnectionImpl.getResponseCode(HttpsURLConnectionImpl.java:334)\tat com.google.api.client.http.javanet.NetHttpResponse.
分享如下代码片段：
public Long processInBatches(Dataset<Row> dataset, StatsDTO statsDTO, EventDTO eventDTO, EcommsDao ecommsDao) throws EcommsDataException {
try {
dfProjectId = properties.getDfprojectId();
cacheDatasetId = properties.getCacheDatasetId();
opsSmtpManageTable = properties.getOpsSmtpManageTable();
tempBucketName = properties.getTemporaryBucket();
int batchSize = 100;
OpsEmailUdfRegister opsEmailUdfRegister = new OpsEmailUdfRegister();
Dataset<Row> enriched = opsEmailUdfRegister.getBodyContent(dataset, ecommsDao);
enriched = enriched.cache();
// Derive number of partitions
long totalRecords = enriched.count();
int numPartitions = (int) Math.ceil((double) totalRecords / batchSize);
// Coalesce to control partitioning and avoid re-computation
enriched = enriched.coalesce(numPartitions);
JavaRDD<Row> rdd = enriched.javaRDD();
JavaRDD<Long> results = rdd.mapPartitions(iterator -> {
List<Long> counts = new ArrayList<>();
List<Row> batch = new ArrayList<>(batchSize);
while (iterator.hasNext()) {
batch.add(iterator.next());
if (batch.size() == batchSize) {
long count = processBodyContent(batch, dfProjectId, cacheDatasetId, opsSmtpManageTable, tempBucketName, statsDTO, eventDTO, ecommsDao);
counts.add(count);
batch.clear();
}
}
if (!batch.isEmpty()) {
long count = processBodyContent(batch, dfProjectId, cacheDatasetId, opsSmtpManageTable, tempBucketName, statsDTO, eventDTO, ecommsDao);
counts.add(count);
}
return counts.iterator();
});
public Long processBodyContent(List<Row> rows, String dfProjectId, String cacheDatasetId, String opsSmtpManageTable, String tempBucketName, StatsDTO statsDTO, EventDTO eventDTO, EcommsDao ecommsDao) throws EcommsDataException {
long errorCount = rows.stream()
.filter(row -> {
String body = row.getAs("BODY");
return ERROR_PROCESSING_EMAIL_BODY.equals(body) || ERROR_INVALID_FILE_FORMAT.equals(body);
}).count();
NON_RELEVANT_COUNT = NON_RELEVANT_COUNT + errorCount;
if (errorCount > 0) {
logger.warn("[OpsEmailSparkPipeline.processBodyContent()] Encountered {} error(s) while processing email body. Current error count is {}", errorCount, NON_RELEVANT_COUNT);
}
long successCount = rows.size() - errorCount;
RELEVANT_COUNT = RELEVANT_COUNT + successCount;
logger.debug("[OpsEmailSparkPipeline.processBodyContent()] Success Count for the batch: {}", successCount);
List<Map<String, Object>> rowsToInsert = new ArrayList<>();
for (Row row : rows) {
Map<String, Object> rowContent = new HashMap<>();
populateRowContent(row, rowContent);
rowContent = ensureRowSizeLimit(rowContent);
rowsToInsert.add(rowContent);
}
insertDataIntoBigQuery(rowsToInsert, dfProjectId, cacheDatasetId, opsSmtpManageTable, tempBucketName, statsDTO, eventDTO);
return (long) rows.size();
}
public void insertDataIntoBigQuery(List<Map<String, Object>> rowsToInsert, String dfProjectId, String cacheDatasetId, String opsSmtpManageTable, String tempBucketName, StatsDTO statsDTO, EventDTO eventDTO) throws EcommsDataException {
logger.info("Entering Inside [OpsEmailSparkPipeline.insertDataIntoBigQuery()]");
TableId tableId = TableId.of(dfProjectId, cacheDatasetId, opsSmtpManageTable);
String tableNameStr = tableId.getProject() + "." + tableId.getDataset() + "." + tableId.getTable();
ObjectMapper mapper = new ObjectMapper();
String filePath = "opsTempFiles/" + UUID.randomUUID() + ".json";
String gcsPath = "gs://" + tempBucketName + "/" + filePath;
Storage storage = StorageOptions.getDefaultInstance().getService();
BlobId blobId = BlobId.of(tempBucketName, filePath);
BlobInfo blobInfo = BlobInfo.newBuilder(blobId).build();
// Compute current UTC timestamp for each bach load for INSERT_DT column
String batchTimestamp = java.time.Instant.now().toString();
// Upload JSON file to GCS bucket
try {
try (WritableByteChannel channel = storage.writer(blobInfo);
BufferedWriter bw = new BufferedWriter(Channels.newWriter(channel, "UTF-8"))) {
for (Map<String, Object> row : rowsToInsert) {
row.put("INSERT_DT", batchTimestamp);
bw.write(mapper.writeValueAsString(row));
bw.newLine();
}
}
logger.info("[OpsEmailSparkPipeline.insertDataIntoBigQuery()] Ops GCS batch load file path: {}", gcsPath);
BigQuery bigQuery = (BigQuery) BigQueryOptions.getDefaultInstance().getService();
LoadJobConfiguration loadConfig =
LoadJobConfiguration.newBuilder(tableId, gcsPath)
.setFormatOptions(FormatOptions.json())
.setWriteDisposition(JobInfo.WriteDisposition.WRITE_APPEND)
.setCreateDisposition(JobInfo.CreateDisposition.CREATE_NEVER)
.build();
// Load data from a GCS json file into the table
Job runningJob = bigQuery.create(JobInfo.of(loadConfig));
JobId jobId = runningJob.getJobId();
logger.info("[OpsEmailSparkPipeline.insertDataIntoBigQuery()] Started BigQuery load job: {}", jobId.getJob());
// Blocks until this load table job completes its execution, with a 2-minute timeout.
Job completedJob;
try {
completedJob = runningJob.waitFor(RetryOption.totalTimeout(Duration.ofMinutes(2)));
} catch (InterruptedException e) {
// Thread.currentThread().interrupt();
throw new EcommsDataException(EcommsErrorCode.SYSTEM_FAILURE, "Ops BigQuery job wait was interrupted for job " + jobId.getJob(), e);
}
if (completedJob == null) {
logger.error("[OpsEmailSparkPipeline.insertDataIntoBigQuery()] BigQuery load job {} timed out after 2 minutes and its status is unknown.", jobId.getJob());
throw new EcommsDataException(EcommsErrorCode.SYSTEM_FAILURE, "Lost track for the Ops BigQuery load job " + jobId.getJob() + " after timeout.");
} else if (completedJob.getStatus().getError() != null) {
logger.error("[OpsEmailSparkPipeline.insertDataIntoBigQuery()] BigQuery load job {} failed for table {}. Error: {}", jobId.getJob(), tableNameStr, completedJob.getStatus().getError());
statsDTO.setNonRelevantCount((long) rowsToInsert.size());
throw new EcommsDataException(EcommsErrorCode.OTHER, "Ops Batch data loading failed into table " + tableNameStr, completedJob.getStatus().getError().toString());
} else {
logger.info(
"[OpsEmailSparkPipeline.insertDataIntoBigQuery()] Successfully loaded {} rows into {}", rowsToInsert.size(), tableNameStr);
}
} catch (BigQueryException | IOException e) {
throw new EcommsDataException(EcommsErrorCode.OTHER, "Batch data loading for Ops failed into table " + tableNameStr, e.getMessage());
} finally {
try {
storage.delete(blobId);
} catch (Exception e) {
logger.warn("[OpsEmailSparkPipeline.insertDataIntoBigQuery()] Failed to delete temporary GCS file: {}. Error: {}", gcsPath, e.getMessage());
}
}
}
解决方案
由于除了使用传统的Spark DataSet写出方式外没有其他可行的变通办法（这会涉及大量现有代码变更），我实现了一个去重逻辑（使用唯一键执行删除的SQL查询），在数据加载操作代码执行完毕后移除重复数据。这样就奏效了。感谢大家的支持。
public static final String GET_DUPLICATE_QUERY = "SELECT COUNT(*) FROM ("
+ "SELECT raw_file_path FROM `%s` WHERE run_date = '%s' AND run_hour = '%s' "
+ "GROUP BY raw_file_path "
+ "HAVING COUNT(*) > 1)";
public static final String DELETE_DUPLICATE_QUERY = "DELETE FROM `%s` "
+ "WHERE run_date = '%s' AND run_hour = '%s' AND (raw_file_path, insert_dt) IN ("
+ "SELECT (raw_file_path, insert_dt) FROM `%s` "
+ "WHERE run_date = '%s' AND run_hour = '%s' "
+ "QUALIFY ROW_NUMBER() OVER (PARTITION BY raw_file_path ORDER BY INSERT_DT DESC) > 1)";
public void deduplicateData(String runDateVal) throws EcommsDataException {
String runDate = runDateVal.substring(0, runDateVal.length() - 2);
String hour = runDateVal.substring(runDateVal.length() - 2);
DateFormat runDateFormat = new SimpleDateFormat(OpsEmailConstants.RUN_DATE_SRC_FORMAT);
DateFormat bqDateFormat = new SimpleDateFormat(OpsEmailConstants.BQ_RUN_DATE_FORMAT);
String dfProjectId = properties.getDfprojectId();
String dfDatasetId = properties.getCacheDatasetId();
String dfSmtpManageTable = properties.getOpsSmtpManageTable();
try {
runDate = bqDateFormat.format(runDateFormat.parse(runDate));
} catch (ParseException e) {
throw new EcommsDataException("[OpsEmailSparkPipeline.deduplicateData()] Exception occurred during date parsing", e);
}
logger.info("[OpsEmailSparkPipeline.deduplicateData()] Starting deduplication for RunDate={}, RunHour={}", runDate, hour);
String dfOpsMngTable = String.format("%s.%s.%s", dfProjectId, dfDatasetId, dfSmtpManageTable);
String getDuplicateQuery = String.format(OpsEmailConstants.GET_DUPLICATE_QUERY, dfOpsMngTable, runDate, hour);
String deleteDuplicateQuery = String.format(OpsEmailConstants.DELETE_DUPLICATE_QUERY, dfOpsMngTable, runDate, hour, dfOpsMngTable, runDate, hour);
String updateCurrTimestampQuery = String.format(OpsEmailConstants.UPDATE_CURR_TIMESTAMP_QUERY, dfOpsMngTable, runDate, hour);
try {
logger.info("[OpsEmailSparkPipeline.deduplicateData()] Executing query to check duplicate records: " + getDuplicateQuery);
BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();
TableResult duplicateCountResult = bigquery.query(QueryJobConfiguration.of(getDuplicateQuery));
long duplicateCount = duplicateCountResult.iterateAll().iterator().next().get(0).getLongValue();
if (duplicateCount > 0) {
logger.info("[OpsEmailSparkPipeline.deduplicateData()] Found {} raw_file_path in table {} with duplicate records " +
"for run_date={} and run_hour={}", duplicateCount, dfOpsMngTable, runDate, hour);
logger.info("[OpsEmailSparkPipeline.deduplicateData()] Executing deduplicate query: " + deleteDuplicateQuery);
bigquery.query(QueryJobConfiguration.of(deleteDuplicateQuery));
logger.info("[OpsEmailSparkPipeline.deduplicateData()] Executing query to update current timestamp: " + updateCurrTimestampQuery);
bigquery.query(QueryJobConfiguration.of(updateCurrTimestampQuery));
} else {
logger.info("[OpsEmailSparkPipeline.deduplicateData()] No duplicate raw_file_path records found in table {} " +
"for run_date={} and run_hour={}", dfOpsMngTable, runDate, hour);
}
} catch (InterruptedException e) {
throw new EcommsDataException("[OpsEmailSparkPipeline.deduplicateData()] Deduplication query execution was interrupted", e);
}
}
}
相关文章
