ADLS integration

SETUP.md

@@ -232,8 +232,9 @@ docker exec -it local.spark-master /bin/bash
 ```
 
 Start `spark-shell` with the following command: Available users are `openhouse` and `u_tableowner`.
+
 ```
-bin/spark-shell --packages org.apache.iceberg:iceberg-spark-runtime-3.1_2.12:1.2.0   \
+bin/spark-shell --packages org.apache.iceberg:iceberg-spark-runtime-3.1_2.12:1.2.0 \
   --jars openhouse-spark-runtime_2.12-*-all.jar  \
   --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions,com.linkedin.openhouse.spark.extensions.OpenhouseSparkSessionExtensions   \
   --conf spark.sql.catalog.openhouse=org.apache.iceberg.spark.SparkCatalog   \
@@ -244,6 +245,23 @@ bin/spark-shell --packages org.apache.iceberg:iceberg-spark-runtime-3.1_2.12:1.2
   --conf spark.sql.catalog.openhouse.cluster=LocalHadoopCluster
 ```
 
+If you are integrating with ADLS, use this `spark-shell` command instead:
+
+```
+bin/spark-shell --packages org.apache.iceberg:iceberg-azure:1.5.0,org.apache.iceberg:iceberg-spark-runtime-3.1_2.12:1.2.0 \
+  --jars openhouse-spark-apps_2.12-*-all.jar,openhouse-spark-runtime_2.12-latest-all.jar  \
+  --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions,com.linkedin.openhouse.spark.extensions.OpenhouseSparkSessionExtensions   \
+  --conf spark.sql.catalog.openhouse=org.apache.iceberg.spark.SparkCatalog   \
+  --conf spark.sql.catalog.openhouse.catalog-impl=com.linkedin.openhouse.spark.OpenHouseCatalog     \
+  --conf spark.sql.catalog.openhouse.metrics-reporter-impl=com.linkedin.openhouse.javaclient.OpenHouseMetricsReporter    \
+  --conf spark.sql.catalog.openhouse.uri=http://openhouse-tables:8080   \
+  --conf spark.sql.catalog.openhouse.auth-token=$(cat /var/config/$(whoami).token) \
+  --conf spark.sql.catalog.openhouse.cluster=LocalABSCluster \
+  --conf spark.sql.catalog.openhouse.io-impl=org.apache.iceberg.azure.adlsv2.ADLSFileIO \
+  --conf spark.sql.catalog.openhouse.adls.auth.shared-key.account.name= <account name> \
+  --conf spark.sql.catalog.openhouse.adls.auth.shared-key.account.key= <account key>
+```
+
 #### Create a table
 
 ```

buildSrc/src/main/groovy/openhouse.iceberg-azure-conventions.gradle

@@ -0,0 +1,21 @@
+ext {
+  icebergAzureVersion = '1.5.2'
+}
+
+dependencies {
+  // Ideally, we have these, but they are only supported for iceberg version >= 1.4.0, which is not compatible
+  // with the current Openhouse implementation.
+  // implementation('org.apache.iceberg:iceberg-azure:' + icebergAzureVersion)
+  // implementation('org.apache.iceberg:iceberg-azure-bundle:' + icebergAzureVersion)
+
+  implementation("com.azure:azure-storage-file-datalake:12.19.1") {
+    exclude group: 'io.netty'
+    exclude group: 'io.projectreactor.netty'
+  }
+
+  implementation("com.azure:azure-identity:1.12.1") {
+    exclude group: 'io.netty'
+    exclude group: 'io.projectreactor.netty'
+  }
+
+}

buildSrc/src/main/groovy/openhouse.iceberg-conventions.gradle

@@ -6,6 +6,6 @@ dependencies {
   implementation('org.apache.iceberg:iceberg-bundled-guava:' + icebergVersion)
   implementation('org.apache.iceberg:iceberg-data:' + icebergVersion)
   implementation('org.apache.iceberg:iceberg-core:' + icebergVersion)
-
-  testImplementation('org.apache.iceberg:iceberg-common:' + icebergVersion)
+  implementation('org.apache.iceberg:iceberg-common:' + icebergVersion)
+  implementation('org.testcontainers:testcontainers:1.19.8')
 }

cluster/storage/build.gradle

@@ -2,11 +2,13 @@ plugins {
   id 'openhouse.java-conventions'
   id 'openhouse.hadoop-conventions'
   id 'openhouse.iceberg-aws-conventions'
+  id 'openhouse.iceberg-azure-conventions'
   id 'openhouse.iceberg-conventions'
   id 'openhouse.maven-publish'
 }
 
 dependencies {
   implementation project(':cluster:configs')
+  implementation project(':iceberg:azure')
   implementation 'org.springframework.boot:spring-boot-autoconfigure:' + spring_web_version
 }

cluster/storage/src/main/java/com/linkedin/openhouse/cluster/storage/StorageType.java

@@ -17,6 +17,7 @@ public class StorageType {
   public static final Type HDFS = new Type("hdfs");
   public static final Type LOCAL = new Type("local");
   public static final Type S3 = new Type("s3");
+  public static final Type ADLS = new Type("adls");
 
   @AllArgsConstructor
   @EqualsAndHashCode
@@ -33,6 +34,8 @@ public Type fromString(String type) {
       return LOCAL;
     } else if (S3.getValue().equals(type)) {
       return S3;
+    } else if (ADLS.getValue().equals(type)) {
+      return ADLS;
     } else {
       throw new IllegalArgumentException("Unknown storage type: " + type);
     }

cluster/storage/src/main/java/com/linkedin/openhouse/cluster/storage/adls/AdlsStorage.java

@@ -0,0 +1,39 @@
+package com.linkedin.openhouse.cluster.storage.adls;
+
+import com.linkedin.openhouse.cluster.storage.BaseStorage;
+import com.linkedin.openhouse.cluster.storage.StorageClient;
+import com.linkedin.openhouse.cluster.storage.StorageType;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.stereotype.Component;
+
+/**
+ * The ADLS Storage class is an implementation of the Storage interface for cloud storage with Azure
+ * Data Lake Storage.
+ */
+@Component
+public class AdlsStorage extends BaseStorage {
+
+  // Declare client class to interact with ADLS filesystems/storage
+  @Autowired @Lazy private AdlsStorageClient adlsStorageClient;
+
+  /**
+   * Get the type of the ADLS storage.
+   *
+   * @return the type of the ADLS storage
+   */
+  @Override
+  public StorageType.Type getType() {
+    return StorageType.ADLS;
+  }
+
+  /**
+   * Get the ADLS storage client.
+   *
+   * @return the ADLS storage client
+   */
+  @Override
+  public StorageClient<?> getClient() {
+    return adlsStorageClient;
+  }
+}

cluster/storage/src/main/java/com/linkedin/openhouse/cluster/storage/adls/AdlsStorageClient.java

@@ -0,0 +1,67 @@
+package com.linkedin.openhouse.cluster.storage.adls;
+
+import com.azure.storage.file.datalake.DataLakeFileClient;
+import com.linkedin.openhouse.cluster.storage.BaseStorageClient;
+import com.linkedin.openhouse.cluster.storage.StorageType;
+import com.linkedin.openhouse.cluster.storage.configs.StorageProperties;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.HashMap;
+import java.util.Map;
+import javax.annotation.PostConstruct;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.stereotype.Component;
+
+/**
+ * AdlsStorageClient is an implementation of the StorageClient interface for Azure Data Lake
+ * Storage. It uses the {@link DataLakeFileClient} to interact with Data Lake Storage.
+ */
+@Slf4j
+@Lazy
+@Component
+public class AdlsStorageClient extends BaseStorageClient<DataLakeFileClient> {
+
+  private static final StorageType.Type ADLS_TYPE = StorageType.ADLS;
+
+  @Autowired private StorageProperties storageProperties;
+
+  private Map properties;
+
+  // TODO: Instantiate DLFC: https://github.com/linkedin/openhouse/issues/148
+  private DataLakeFileClient dataLakeClient = null;
+
+  /** Intialize the ADLS Client when the bean is accessed the first time. */
+  @PostConstruct
+  public synchronized void init() {
+
+    validateProperties();
+
+    // Gets the parameters from the ADLS storage type
+    properties =
+        new HashMap(storageProperties.getTypes().get(ADLS_TYPE.getValue()).getParameters());
+
+    // Try to create a URI with the endpoint and rootpath
+    URI uri;
+    try {
+      uri = new URI(getEndpoint() + getRootPrefix());
+    } catch (URISyntaxException e) {
+      throw new IllegalArgumentException(
+          String.format(
+              "Bad storage properties provided: [endpoint: %s, rootPrefix: %s] for type [%s]",
+              getEndpoint(), getRootPrefix(), ADLS_TYPE.getValue()),
+          e);
+    }
+  }
+
+  @Override
+  public DataLakeFileClient getNativeClient() {
+    return dataLakeClient;
+  }
+
+  @Override
+  public StorageType.Type getStorageType() {
+    return ADLS_TYPE;
+  }
+}