diff --git a/docs/usage/properties/instance/user/bulk_import.md b/docs/usage/properties/instance/user/bulk_import.md
index 9fefa86e082..4b54f04e79a 100644
--- a/docs/usage/properties/instance/user/bulk_import.md
+++ b/docs/usage/properties/instance/user/bulk_import.md
@@ -11,6 +11,7 @@ Note that on EMR, the total resource allocation must align with the instance typ
| sleeper.bulk.import.emr.spark.speculation | If true then speculative execution of tasks will be performed. Used to set spark.speculation.
See https://spark.apache.org/docs/latest/configuration.html. | false | true |
| sleeper.bulk.import.spark.speculation.quantile | Fraction of tasks which must be complete before speculation is enabled for a particular stage. Used to set spark.speculation.quantile.
See https://spark.apache.org/docs/latest/configuration.html. | 0.75 | true |
| sleeper.bulk.import.starter.memory.mb | The amount of memory in MB for lambda functions that start bulk import jobs. | | true |
+| sleeper.bulk.import.job.file.retention.days | The number of days a bulk import job is held in the bulk import bucket. When a job is submitted to Spark it is written to S3 for Spark to read. This is retained in case the job needs to be retried, and for diagnostic purposes.
Defaults to the value of sleeper.log.retention.days. | | true |
| sleeper.bulk.import.emr.spark.executor.memory | The amount of memory allocated to a Spark executor. Used to set spark.executor.memory.
See https://spark.apache.org/docs/latest/configuration.html. | 16g | true |
| sleeper.bulk.import.emr.spark.driver.memory | The amount of memory allocated to the Spark driver. Used to set spark.driver.memory.
See https://spark.apache.org/docs/latest/configuration.html. | 16g | true |
| sleeper.bulk.import.emr.spark.executor.instances | The number of executors. Used to set spark.executor.instances.
See https://spark.apache.org/docs/latest/configuration.html. | 29 | true |
diff --git a/example/full/instance.properties b/example/full/instance.properties
index b65bed88aa0..c160bc8bf06 100644
--- a/example/full/instance.properties
+++ b/example/full/instance.properties
@@ -677,6 +677,13 @@
# (default value shown below, uncomment to set a value)
# sleeper.bulk.import.starter.memory.mb=4096
+# The number of days a bulk import job is held in the bulk import bucket. When a job is submitted to
+# Spark it is written to S3 for Spark to read. This is retained in case the job needs to be retried,
+# and for diagnostic purposes.
+# Defaults to the value of sleeper.log.retention.days.
+# (default value shown below, uncomment to set a value)
+# sleeper.bulk.import.job.file.retention.days=30
+
# The amount of memory allocated to a Spark executor. Used to set spark.executor.memory.
# See https://spark.apache.org/docs/latest/configuration.html.
# (default value shown below, uncomment to set a value)
diff --git a/java/bulk-import/bulk-import-core/src/main/java/sleeper/bulkimport/core/job/BulkImportJob.java b/java/bulk-import/bulk-import-core/src/main/java/sleeper/bulkimport/core/job/BulkImportJob.java
index 6819a0afd4e..4b9395fa669 100644
--- a/java/bulk-import/bulk-import-core/src/main/java/sleeper/bulkimport/core/job/BulkImportJob.java
+++ b/java/bulk-import/bulk-import-core/src/main/java/sleeper/bulkimport/core/job/BulkImportJob.java
@@ -26,6 +26,12 @@
* POJO containing information needed to run a bulk import job.
*/
public class BulkImportJob {
+
+ /**
+ * The prefix for files holding a bulk import job in the bulk import bucket.
+ */
+ public static final String FILES_BUCKET_PREFIX = "bulk_import/";
+
private final String id;
private final String tableName;
private final String tableId;
diff --git a/java/bulk-import/bulk-import-runner/src/main/java/sleeper/bulkimport/runner/BulkImportJobLoaderFromS3.java b/java/bulk-import/bulk-import-runner/src/main/java/sleeper/bulkimport/runner/BulkImportJobLoaderFromS3.java
index 90036409261..70b417f9a2a 100644
--- a/java/bulk-import/bulk-import-runner/src/main/java/sleeper/bulkimport/runner/BulkImportJobLoaderFromS3.java
+++ b/java/bulk-import/bulk-import-runner/src/main/java/sleeper/bulkimport/runner/BulkImportJobLoaderFromS3.java
@@ -15,11 +15,9 @@
*/
package sleeper.bulkimport.runner;
-import com.google.gson.JsonSyntaxException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import software.amazon.awssdk.services.s3.S3Client;
-import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
import software.amazon.awssdk.services.s3.model.GetObjectRequest;
import sleeper.bulkimport.core.job.BulkImportJob;
@@ -46,16 +44,6 @@ public static BulkImportJob loadJob(InstanceProperties instanceProperties, Strin
.bucket(bulkImportBucket)
.key(objectKey)
.build()).asUtf8String();
- try {
- return new BulkImportJobSerDe().fromJson(jsonJob);
- } catch (JsonSyntaxException e) {
- LOGGER.error("Json job was malformed");
- throw e;
- } finally {
- s3Client.deleteObject(DeleteObjectRequest.builder()
- .bucket(bulkImportBucket)
- .key(objectKey)
- .build());
- }
+ return new BulkImportJobSerDe().fromJson(jsonJob);
}
}
diff --git a/java/bulk-import/bulk-import-runner/src/test/java/sleeper/bulkimport/runner/BulkImportJobLoaderFromS3IT.java b/java/bulk-import/bulk-import-runner/src/test/java/sleeper/bulkimport/runner/BulkImportJobLoaderFromS3IT.java
index fa83195c4f1..53ad4575c43 100644
--- a/java/bulk-import/bulk-import-runner/src/test/java/sleeper/bulkimport/runner/BulkImportJobLoaderFromS3IT.java
+++ b/java/bulk-import/bulk-import-runner/src/test/java/sleeper/bulkimport/runner/BulkImportJobLoaderFromS3IT.java
@@ -51,13 +51,21 @@ void shouldLoadBulkImportJobFromS3() {
.files(List.of("/load-job.parquet"))
.build();
- BulkImportJobWriterToS3 bulkImportJobWriterToS3 = new BulkImportJobWriterToS3(instanceProperties, s3Client);
- bulkImportJobWriterToS3.writeJobToBulkImportBucket(bulkImportJob, objectKey);
-
- // When / Then
- assertThat(BulkImportJobLoaderFromS3.loadJob(instanceProperties, objectKey, s3Client))
- .isEqualTo(bulkImportJob);
- // And the file is deleted after it is loaded
- assertThat(listObjectKeys(instanceProperties.get(BULK_IMPORT_BUCKET))).isEmpty();
+ // When
+ writer().writeJobToBulkImportBucket(bulkImportJob, objectKey);
+ BulkImportJob foundJob = loadJob(objectKey);
+
+ // Then
+ assertThat(foundJob).isEqualTo(bulkImportJob);
+ // And the file is kept
+ assertThat(listObjectKeys(instanceProperties.get(BULK_IMPORT_BUCKET))).containsExactly(objectKey);
+ }
+
+ private BulkImportJobWriterToS3 writer() {
+ return new BulkImportJobWriterToS3(instanceProperties, s3Client);
+ }
+
+ private BulkImportJob loadJob(String objectKey) {
+ return BulkImportJobLoaderFromS3.loadJob(instanceProperties, objectKey, s3Client);
}
}
diff --git a/java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/executor/BulkImportExecutor.java b/java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/executor/BulkImportExecutor.java
index 9fd3240fbb4..e9b7a7a57f2 100644
--- a/java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/executor/BulkImportExecutor.java
+++ b/java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/executor/BulkImportExecutor.java
@@ -58,7 +58,7 @@ public BulkImportExecutor(
}
public static String createJobFileObjectKey(BulkImportJob job, String jobRunId) {
- return "bulk_import/" + job.getId() + "-" + jobRunId + ".json";
+ return BulkImportJob.FILES_BUCKET_PREFIX + job.getId() + "-" + jobRunId + ".json";
}
public void runJob(BulkImportJob bulkImportJob) {
diff --git a/java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/executor/BulkImportJobWriterToS3.java b/java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/executor/BulkImportJobWriterToS3.java
index 4cef05283bb..a0daaa8a95a 100644
--- a/java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/executor/BulkImportJobWriterToS3.java
+++ b/java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/executor/BulkImportJobWriterToS3.java
@@ -30,8 +30,8 @@
public class BulkImportJobWriterToS3 implements BulkImportExecutor.WriteJobToBucket {
public static final Logger LOGGER = LoggerFactory.getLogger(BulkImportJobWriterToS3.class);
- protected final InstanceProperties instanceProperties;
- protected final S3Client s3Client;
+ private final InstanceProperties instanceProperties;
+ private final S3Client s3Client;
public BulkImportJobWriterToS3(InstanceProperties instanceProperties, S3Client s3Client) {
this.instanceProperties = instanceProperties;
diff --git a/java/core/src/main/java/sleeper/core/properties/instance/BulkImportProperty.java b/java/core/src/main/java/sleeper/core/properties/instance/BulkImportProperty.java
index bb2419a9337..81c8f636777 100644
--- a/java/core/src/main/java/sleeper/core/properties/instance/BulkImportProperty.java
+++ b/java/core/src/main/java/sleeper/core/properties/instance/BulkImportProperty.java
@@ -21,6 +21,7 @@
import java.util.List;
+import static sleeper.core.properties.instance.CommonProperty.LOG_RETENTION_IN_DAYS;
import static sleeper.core.properties.instance.TableStateProperty.DEFAULT_TABLE_STATE_LAMBDA_MEMORY;
/**
@@ -57,6 +58,15 @@ public interface BulkImportProperty {
.defaultProperty(DEFAULT_TABLE_STATE_LAMBDA_MEMORY)
.propertyGroup(InstancePropertyGroup.BULK_IMPORT)
.runCdkDeployWhenChanged(true).build();
+ UserDefinedInstanceProperty BULK_IMPORT_JOB_FILE_RETENTION_DAYS = Index.propertyBuilder("sleeper.bulk.import.job.file.retention.days")
+ .description("The number of days a bulk import job is held in the bulk import bucket. When a job is " +
+ "submitted to Spark it is written to S3 for Spark to read. This is retained in case the job " +
+ "needs to be retried, and for diagnostic purposes.\n" +
+ "Defaults to the value of sleeper.log.retention.days.")
+ .defaultProperty(LOG_RETENTION_IN_DAYS)
+ .validationPredicate(SleeperPropertyValueUtils::isPositiveInteger)
+ .propertyGroup(InstancePropertyGroup.BULK_IMPORT)
+ .runCdkDeployWhenChanged(true).build();
static List getAll() {
return Index.INSTANCE.getAll();
diff --git a/java/core/src/test/java/sleeper/core/properties/SleeperPropertiesValidationTest.java b/java/core/src/test/java/sleeper/core/properties/SleeperPropertiesValidationTest.java
index 88d142e8674..62d55acaf24 100644
--- a/java/core/src/test/java/sleeper/core/properties/SleeperPropertiesValidationTest.java
+++ b/java/core/src/test/java/sleeper/core/properties/SleeperPropertiesValidationTest.java
@@ -29,10 +29,10 @@
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static sleeper.core.properties.PropertiesUtils.loadProperties;
+import static sleeper.core.properties.instance.BulkImportProperty.BULK_IMPORT_JOB_FILE_RETENTION_DAYS;
import static sleeper.core.properties.instance.CdkDefinedInstanceProperty.VERSION;
import static sleeper.core.properties.instance.CommonProperty.ID;
import static sleeper.core.properties.instance.CommonProperty.JARS_BUCKET;
-import static sleeper.core.properties.instance.CommonProperty.LOG_RETENTION_IN_DAYS;
import static sleeper.core.properties.instance.CommonProperty.MAXIMUM_CONNECTIONS_TO_S3;
import static sleeper.core.properties.instance.CommonProperty.SUBNETS;
import static sleeper.core.properties.instance.CommonProperty.VPC_ID;
@@ -168,17 +168,17 @@ class MultipleValidationErrors {
void shouldFailValidationWithTwoInvalidInstanceProperties() {
// Given
InstanceProperties instanceProperties = createTestInstanceProperties();
- instanceProperties.set(LOG_RETENTION_IN_DAYS, "abc");
+ instanceProperties.set(BULK_IMPORT_JOB_FILE_RETENTION_DAYS, "abc");
instanceProperties.set(MAXIMUM_CONNECTIONS_TO_S3, "def");
// When
assertThatThrownBy(instanceProperties::validate)
.isInstanceOf(SleeperPropertiesInvalidException.class)
- .hasMessageContaining("Property sleeper.log.retention.days was invalid. It was \"abc\".")
+ .hasMessageContaining("Property sleeper.fs.s3a.max-connections was invalid. It was \"def\".")
.hasMessageContaining("Failure 1 of 2.")
.extracting("invalidValues")
.isEqualTo(Map.of(
- LOG_RETENTION_IN_DAYS, "abc",
+ BULK_IMPORT_JOB_FILE_RETENTION_DAYS, "abc",
MAXIMUM_CONNECTIONS_TO_S3, "def"));
}
diff --git a/java/deployment/cdk/src/main/java/sleeper/cdk/stack/bulkimport/BulkImportBucketStack.java b/java/deployment/cdk/src/main/java/sleeper/cdk/stack/bulkimport/BulkImportBucketStack.java
index 272785ec539..c02085c3186 100644
--- a/java/deployment/cdk/src/main/java/sleeper/cdk/stack/bulkimport/BulkImportBucketStack.java
+++ b/java/deployment/cdk/src/main/java/sleeper/cdk/stack/bulkimport/BulkImportBucketStack.java
@@ -15,19 +15,26 @@
*/
package sleeper.cdk.stack.bulkimport;
+import software.amazon.awscdk.Duration;
import software.amazon.awscdk.NestedStack;
import software.amazon.awscdk.RemovalPolicy;
import software.amazon.awscdk.services.s3.BlockPublicAccess;
import software.amazon.awscdk.services.s3.Bucket;
import software.amazon.awscdk.services.s3.BucketEncryption;
import software.amazon.awscdk.services.s3.IBucket;
+import software.amazon.awscdk.services.s3.LifecycleRule;
import software.constructs.Construct;
+import sleeper.bulkimport.core.job.BulkImportJob;
import sleeper.cdk.stack.SleeperCoreStacks;
import sleeper.cdk.util.S3BucketName;
import sleeper.core.properties.instance.InstanceProperties;
+import java.util.List;
+
+import static sleeper.core.properties.instance.BulkImportProperty.BULK_IMPORT_JOB_FILE_RETENTION_DAYS;
import static sleeper.core.properties.instance.CdkDefinedInstanceProperty.BULK_IMPORT_BUCKET;
+import static sleeper.core.properties.instance.CommonProperty.LOG_RETENTION_IN_DAYS;
public class BulkImportBucketStack extends NestedStack {
private final IBucket importBucket;
@@ -42,6 +49,19 @@ public BulkImportBucketStack(Construct scope, String id, InstanceProperties inst
.versioned(false)
.removalPolicy(RemovalPolicy.DESTROY)
.encryption(BucketEncryption.S3_MANAGED)
+ .lifecycleRules(List.of(
+ LifecycleRule.builder()
+ .prefix(BulkImportJob.FILES_BUCKET_PREFIX)
+ .expiration(Duration.days(instanceProperties.getInt(BULK_IMPORT_JOB_FILE_RETENTION_DAYS)))
+ .build(),
+ LifecycleRule.builder()
+ .prefix("logs/")
+ .expiration(Duration.days(instanceProperties.getInt(LOG_RETENTION_IN_DAYS)))
+ .build(),
+ LifecycleRule.builder()
+ .prefix("applications/")
+ .expiration(Duration.days(instanceProperties.getInt(LOG_RETENTION_IN_DAYS)))
+ .build()))
.build();
importBucket.grantWrite(coreStacks.getIngestByQueuePolicyForGrants());
instanceProperties.set(BULK_IMPORT_BUCKET, importBucket.getBucketName());
diff --git a/java/deployment/cdk/src/test/java/sleeper/cdk/default-instance.approved.json b/java/deployment/cdk/src/test/java/sleeper/cdk/default-instance.approved.json
index e01126d64d7..19e29caa46d 100644
--- a/java/deployment/cdk/src/test/java/sleeper/cdk/default-instance.approved.json
+++ b/java/deployment/cdk/src/test/java/sleeper/cdk/default-instance.approved.json
@@ -13210,6 +13210,25 @@
}
}
]
+ },
+ "LifecycleConfiguration": {
+ "Rules": [
+ {
+ "ExpirationInDays": 30,
+ "Prefix": "bulk_import/",
+ "Status": "Enabled"
+ },
+ {
+ "ExpirationInDays": 30,
+ "Prefix": "logs/",
+ "Status": "Enabled"
+ },
+ {
+ "ExpirationInDays": 30,
+ "Prefix": "applications/",
+ "Status": "Enabled"
+ }
+ ]
}
}
},
diff --git a/scripts/templates/instanceproperties.template b/scripts/templates/instanceproperties.template
index e30c8fe77ac..c836c96a5c6 100644
--- a/scripts/templates/instanceproperties.template
+++ b/scripts/templates/instanceproperties.template
@@ -681,6 +681,13 @@
# (default value shown below, uncomment to set a value)
# sleeper.bulk.import.starter.memory.mb=4096
+# The number of days a bulk import job is held in the bulk import bucket. When a job is submitted to
+# Spark it is written to S3 for Spark to read. This is retained in case the job needs to be retried,
+# and for diagnostic purposes.
+# Defaults to the value of sleeper.log.retention.days.
+# (default value shown below, uncomment to set a value)
+# sleeper.bulk.import.job.file.retention.days=30
+
# The amount of memory allocated to a Spark executor. Used to set spark.executor.memory.
# See https://spark.apache.org/docs/latest/configuration.html.
# (default value shown below, uncomment to set a value)