-
Notifications
You must be signed in to change notification settings - Fork 1.7k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
b1b84d5
commit 99bc634
Showing
9 changed files
with
1,153 additions
and
39 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
681 changes: 681 additions & 0 deletions
681
spark/src/main/java/io/delta/dynamodbcommitstore/DynamoDBCommitOwnerClient.java
Large diffs are not rendered by default.
Oops, something went wrong.
72 changes: 72 additions & 0 deletions
72
spark/src/main/java/io/delta/dynamodbcommitstore/DynamoDBCommitOwnerClientBuilder.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| /* | ||
| * Copyright (2021) The Delta Lake Project Authors. | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package io.delta.dynamodbcommitstore; | ||
|
|
||
| import com.amazonaws.auth.AWSCredentialsProvider; | ||
| import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient; | ||
| import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClientBuilder; | ||
| import org.apache.spark.sql.delta.managedcommit.CommitOwnerBuilder; | ||
| import org.apache.spark.sql.delta.managedcommit.CommitOwnerClient; | ||
| import scala.collection.immutable.Map; | ||
|
|
||
| import java.lang.reflect.InvocationTargetException; | ||
|
|
||
| public class DynamoDBCommitOwnerClientBuilder implements CommitOwnerBuilder { | ||
|
|
||
| private final long BACKFILL_BATCH_SIZE = 1L; | ||
|
|
||
| @Override | ||
| public String getName() { | ||
| return "dynamodb"; | ||
| } | ||
|
|
||
| private static final String MANAGED_COMMITS_TABLE_NAME_KEY = "managedCommitsTableName"; | ||
| private static final String DYNAMO_DB_ENDPOINT_KEY = "dynamoDBEndpoint"; | ||
| private static final String AWS_CREDENTIALS_PROVIDER_KEY = "awsCredentialsProvider"; | ||
|
|
||
| // TODO: update this interface so that it can take a sparkSession. | ||
| @Override | ||
| public CommitOwnerClient build(Map<String, String> conf) { | ||
| String managedCommitsTableName = conf.get(MANAGED_COMMITS_TABLE_NAME_KEY).getOrElse(() -> { | ||
| throw new RuntimeException(MANAGED_COMMITS_TABLE_NAME_KEY + " not found"); | ||
| }); | ||
| String dynamoDBEndpoint = conf.get(DYNAMO_DB_ENDPOINT_KEY).getOrElse(() -> { | ||
| throw new RuntimeException(DYNAMO_DB_ENDPOINT_KEY + " not found"); | ||
| }); | ||
| String awsCredentialsProviderName = conf.get(AWS_CREDENTIALS_PROVIDER_KEY).getOrElse( () -> | ||
| "com.amazonaws.auth.DefaultAWSCredentialsProviderChain"); | ||
| try { | ||
| AmazonDynamoDBClient client = | ||
| createAmazonDDBClient(dynamoDBEndpoint, awsCredentialsProviderName); | ||
| return new DynamoDBCommitOwnerClient( | ||
| managedCommitsTableName, dynamoDBEndpoint, client, BACKFILL_BATCH_SIZE); | ||
| } catch (Exception e) { | ||
| throw new RuntimeException("Failed to create DynamoDB client", e); | ||
| } | ||
| } | ||
|
|
||
| private AmazonDynamoDBClient createAmazonDDBClient( | ||
| String endpoint, | ||
| String credentialProviderName) throws NoSuchMethodException, ClassNotFoundException, InvocationTargetException, InstantiationException, IllegalAccessException { | ||
| Class<?> awsCredentialsProviderClass = Class.forName(credentialProviderName); | ||
| AWSCredentialsProvider awsCredentialsProvider = | ||
| (AWSCredentialsProvider) awsCredentialsProviderClass.getConstructor().newInstance(); | ||
| AmazonDynamoDBClient client = new AmazonDynamoDBClient(awsCredentialsProvider); | ||
| client.setEndpoint(endpoint); | ||
| return client; | ||
| } | ||
| } |
49 changes: 49 additions & 0 deletions
49
spark/src/main/java/io/delta/dynamodbcommitstore/DynamoDBTableEntryConstants.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,49 @@ | ||
| /* | ||
| * Copyright (2021) The Delta Lake Project Authors. | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package io.delta.dynamodbcommitstore; | ||
|
|
||
| /** | ||
| * Defines the field names used in the DynamoDB table entry. | ||
| */ | ||
| final class DynamoDBTableEntryConstants { | ||
| private DynamoDBTableEntryConstants() {} | ||
|
|
||
| /** The primary key of the DynamoDB table. */ | ||
| public static final String TABLE_ID = "tableId"; | ||
| /** The version of the latest commit in the corresponding Delta table. */ | ||
| public static final String TABLE_LATEST_VERSION = "tableVersion"; | ||
| /** The inCommitTimestamp of the latest commit in the corresponding Delta table. */ | ||
| public static final String TABLE_LATEST_TIMESTAMP = "tableTimestamp"; | ||
| /** Whether this commit owner is accepting more commits for the corresponding Delta table. */ | ||
| public static final String ACCEPTING_COMMITS = "acceptingCommits"; | ||
| /** The path of the corresponding Delta table. */ | ||
| public static final String TABLE_PATH = "path"; | ||
| /** The schema version of this DynamoDB table entry. */ | ||
| public static final String SCHEMA_VERSION = "schemaVersion"; | ||
| /** The name of the field used to store unbackfilled commits. */ | ||
| public static final String COMMITS = "commits"; | ||
| /** The unbackfilled commit version. */ | ||
| public static final String COMMIT_VERSION = "version"; | ||
| /** The inCommitTimestamp of the unbackfilled commit. */ | ||
| public static final String COMMIT_TIMESTAMP = "timestamp"; | ||
| /** The name of the unbackfilled file. e.g. 00001.uuid.json */ | ||
| public static final String COMMIT_FILE_NAME = "fsName"; | ||
| /** The length of the unbackfilled file as per the file status. */ | ||
| public static final String COMMIT_FILE_LENGTH = "fsLength"; | ||
| /** The modification timestamp of the unbackfilled file as per the file status. */ | ||
| public static final String COMMIT_FILE_MODIFICATION_TIMESTAMP = "fsTimestamp"; | ||
| } |
77 changes: 77 additions & 0 deletions
77
spark/src/main/java/io/delta/dynamodbcommitstore/ManagedCommitUtils.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,77 @@ | ||
| /* | ||
| * Copyright (2021) The Delta Lake Project Authors. | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package io.delta.dynamodbcommitstore; | ||
|
|
||
| import org.apache.spark.sql.delta.managedcommit.AbstractMetadata; | ||
| import org.apache.spark.sql.delta.managedcommit.UpdatedActions; | ||
| import org.apache.hadoop.fs.Path; | ||
|
|
||
| import java.util.UUID; | ||
|
|
||
| public class ManagedCommitUtils { | ||
|
|
||
| private ManagedCommitUtils() {} | ||
|
|
||
| /** The subdirectory in which to store the unbackfilled commit files. */ | ||
| final static String COMMIT_SUBDIR = "_commits"; | ||
|
|
||
| /** The configuration key for the managed commit owner. */ | ||
| private static final String MANAGED_COMMIT_OWNER_CONF_KEY = | ||
| "delta.managedCommits.commitOwner-dev"; | ||
|
|
||
| /** | ||
| * Creates a new unbackfilled delta file path for the given commit version. | ||
| * The path is of the form `tablePath/_delta_log/_commits/00000000000000000001.uuid.json`. | ||
| */ | ||
| public static Path generateUnbackfilledDeltaFilePath( | ||
| Path logPath, | ||
| long version) { | ||
| String uuid = UUID.randomUUID().toString(); | ||
| Path basePath = new Path(logPath, COMMIT_SUBDIR); | ||
| return new Path(basePath, String.format("%020d.%s.json", version, uuid)); | ||
| } | ||
|
|
||
| /** | ||
| * Returns the path to the backfilled delta file for the given commit version. | ||
| * The path is of the form `tablePath/_delta_log/00000000000000000001.json`. | ||
| */ | ||
| public static Path getBackfilledDeltaFilePath( | ||
| Path logPath, | ||
| Long version) { | ||
| return new Path(logPath, String.format("%020d.json", version)); | ||
| } | ||
|
|
||
| private static String getManagedCommitOwner(AbstractMetadata metadata) { | ||
| return metadata | ||
| .getConfiguration() | ||
| .get(MANAGED_COMMIT_OWNER_CONF_KEY) | ||
| .getOrElse(() -> ""); | ||
| } | ||
|
|
||
| /** | ||
| * Returns true if the commit is a managed commit to filesystem conversion. | ||
| */ | ||
| public static boolean isManagedCommitToFSConversion( | ||
| Long commitVersion, | ||
| UpdatedActions updatedActions) { | ||
| boolean oldMetadataHasManagedCommits = | ||
| getManagedCommitOwner(updatedActions.getOldMetadata()).isEmpty(); | ||
| boolean newMetadataHasManagedCommits = | ||
| getManagedCommitOwner(updatedActions.getNewMetadata()).isEmpty(); | ||
| return oldMetadataHasManagedCommits && !newMetadataHasManagedCommits && commitVersion > 0; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.