Merge branch 'master' into FSTORE-1023

Author: bubriks

.github/workflows/python-lint.yml

@@ -11,7 +11,7 @@ jobs:
       - uses: actions/checkout@v3
       - uses: actions/setup-python@v4
         with:
-          python-version: '3.8'
+          python-version: '3.10'
       - name: install deps
         run: pip install flake8==3.9.0 black==22.3.0 pre-commit-hooks==2.4.0
 
@@ -33,7 +33,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.7", "3.8", "3.9", "3.10"]
+        python-version: ["3.8", "3.9", "3.10"]
 
     steps:
       - name: Set Timezone
@@ -55,6 +55,8 @@ jobs:
         run: python --version
 
       - name: Run Pytest suite
+        env:
+          ENABLE_HOPSWORKS_USAGE: "false"
         run: pytest python/tests
 
   unit_tests_ubuntu_pandas:
@@ -73,7 +75,7 @@ jobs:
       - uses: actions/setup-python@v4
         name: Setup Python
         with:
-          python-version: "3.9"
+          python-version: "3.10"
           cache: "pip"
           cache-dependency-path: "python/setup.py"
       - run: pip install -e python[python,dev-pandas1,docs]
@@ -82,6 +84,8 @@ jobs:
         run: python --version
 
       - name: Run Pytest suite
+        env:
+          ENABLE_HOPSWORKS_USAGE: "false"
         run: pytest python/tests
 
   unit_tests_ubuntu_local:
@@ -100,7 +104,7 @@ jobs:
       - uses: actions/setup-python@v4
         name: Setup Python
         with:
-          python-version: "3.9"
+          python-version: "3.10"
           cache: "pip"
           cache-dependency-path: "python/setup.py"
       - run: pip install -e python[python,dev,docs]
@@ -109,6 +113,8 @@ jobs:
         run: python --version
 
       - name: Run Pytest suite
+        env:
+          ENABLE_HOPSWORKS_USAGE: "false"
         run: pytest python/tests
 
   unit_tests_windows_utc:
@@ -127,7 +133,7 @@ jobs:
       - uses: actions/setup-python@v4
         name: Setup Python
         with:
-          python-version: "3.9"
+          python-version: "3.10"
           cache: "pip"
           cache-dependency-path: "python/setup.py"
       - run: pip install -e python[python,dev,docs]
@@ -136,6 +142,8 @@ jobs:
         run: python --version
 
       - name: Run Pytest suite
+        env:
+          ENABLE_HOPSWORKS_USAGE: "false"
         run: pytest python/tests
 
 
@@ -155,7 +163,7 @@ jobs:
       - uses: actions/setup-python@v4
         name: Setup Python
         with:
-          python-version: "3.9"
+          python-version: "3.10"
           cache: "pip"
           cache-dependency-path: "python/setup.py"
       - run: pip install -e python[python,dev,docs]
@@ -164,4 +172,6 @@ jobs:
         run: python --version
 
       - name: Run Pytest suite
+        env:
+          ENABLE_HOPSWORKS_USAGE: "false"
         run: pytest python/tests

java/beam/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <artifactId>hsfs-parent</artifactId>
     <groupId>com.logicalclocks</groupId>
-    <version>3.5.0-SNAPSHOT</version>
+    <version>3.7.0-SNAPSHOT</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
 

java/flink/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <artifactId>hsfs-parent</artifactId>
     <groupId>com.logicalclocks</groupId>
-    <version>3.5.0-SNAPSHOT</version>
+    <version>3.7.0-SNAPSHOT</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
 

java/hsfs/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <artifactId>hsfs-parent</artifactId>
     <groupId>com.logicalclocks</groupId>
-    <version>3.5.0-SNAPSHOT</version>
+    <version>3.7.0-SNAPSHOT</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
 

java/pom.xml

@@ -7,7 +7,7 @@
     <groupId>com.logicalclocks</groupId>
     <artifactId>hsfs-parent</artifactId>
     <packaging>pom</packaging>
-    <version>3.5.0-SNAPSHOT</version>
+    <version>3.7.0-SNAPSHOT</version>
     <modules>
         <module>hsfs</module>
         <module>spark</module>
@@ -27,18 +27,20 @@
         <lombok.version>1.18.10</lombok.version>
         <fasterxml.jackson.databind.version>2.10.0</fasterxml.jackson.databind.version>
         <deequ.version>1.1.0.6-SNAPSHOT</deequ.version>
-        <hudi.version>0.10.0.3</hudi.version>
+        <hudi.version>0.12.3.0</hudi.version>
         <awssdk.version>2.10.40</awssdk.version>
         <scala.version>2.12.10</scala.version>
         <scala-short.version>2.12</scala-short.version>
         <dbutils.version>0.0.5</dbutils.version>
-        <json.version>20230227</json.version>
+        <json.version>20231013</json.version>
         <hoverfly.version>0.12.2</hoverfly.version>
         <junit.version>5.9.1</junit.version>
         <surefire-plugin.version>2.22.0</surefire-plugin.version>
         <mockito.version>4.3.1</mockito.version>
         <avro.version>1.8.2</avro.version>
 
+        <artifact.spark.version>spark3.1</artifact.spark.version>
+
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <delombok.output>${project.basedir}/delombok</delombok.output>
     </properties>
@@ -271,6 +273,17 @@
             </testResource>
         </testResources>
     </build>
+
+    <profiles>
+        <profile>
+            <id>spark-3.3</id>
+            <properties>
+                <deequ.version>2.0.4.0-spark-3.3</deequ.version>
+                <artifact.spark.version>spark3.3</artifact.spark.version>
+            </properties>
+        </profile>
+    </profiles>
+
     <repositories>
         <repository>
             <id>Hops</id>

java/spark/pom.xml

@@ -22,11 +22,11 @@
     <parent>
         <artifactId>hsfs-parent</artifactId>
         <groupId>com.logicalclocks</groupId>
-        <version>3.5.0-SNAPSHOT</version>
+        <version>3.7.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
-    <artifactId>hsfs-spark</artifactId>
+    <artifactId>hsfs-spark-${artifact.spark.version}</artifactId>
 
     <properties>
         <spark.version>3.1.1.3</spark.version>
@@ -123,7 +123,7 @@
         <!-- https://mvnrepository.com/artifact/org.apache.hudi/hudi-spark-bundle -->
         <dependency>
             <groupId>io.hops.hudi</groupId>
-            <artifactId>hudi-spark3-bundle_${scala-short.version}</artifactId>
+            <artifactId>hudi-spark3.1-bundle_${scala-short.version}</artifactId>
             <version>${hudi.version}</version>
             <scope>provided</scope>
             <exclusions>

java/spark/src/main/java/com/logicalclocks/hsfs/spark/engine/SparkEngine.java

@@ -577,6 +577,7 @@ public <S> StreamingQuery writeStreamDataframe(FeatureGroupBase featureGroupBase
     byte[] featureGroupId = String.valueOf(featureGroupBase.getId()).getBytes(StandardCharsets.UTF_8);
     byte[] schemaId = String.valueOf(featureGroupBase.getSubject().getSchemaId()).getBytes(StandardCharsets.UTF_8);
 
+    queryName = makeQueryName(queryName, featureGroupBase);
     DataStreamWriter<Row> writer =
         onlineFeatureGroupToAvro(featureGroupBase, encodeComplexFeatures(featureGroupBase, dataset))
             .withColumn("headers", array(
@@ -596,8 +597,9 @@ public <S> StreamingQuery writeStreamDataframe(FeatureGroupBase featureGroupBase
             .writeStream()
             .format(Constants.KAFKA_FORMAT)
             .outputMode(outputMode)
+            .queryName(queryName)
             .option("checkpointLocation", checkpointLocation == null
-                ? checkpointDirPath(queryName, featureGroupBase.getOnlineTopicName())
+                ? checkpointDirPath(queryName)
                 : checkpointLocation)
             .options(writeOptions)
             .option("topic", featureGroupBase.getOnlineTopicName());
@@ -1082,12 +1084,21 @@ public String constructCheckpointPath(FeatureGroupBase featureGroup, String quer
         + "/Resources/" + queryName + "-checkpoint";
   }
 
-  public String checkpointDirPath(String queryName, String onlineTopicName) throws FeatureStoreException {
-    if (Strings.isNullOrEmpty(queryName)) {
-      queryName = "insert_stream_" + onlineTopicName;
-    }
+  private String checkpointDirPath(String queryName) throws FeatureStoreException {
     return "/Projects/" + HopsworksClient.getInstance().getProject().getProjectName()
         + "/Resources/" + queryName + "-checkpoint";
   }
 
+  protected String makeQueryName(String queryName, FeatureGroupBase featureGroup) {
+    if (Strings.isNullOrEmpty(queryName)) {
+      queryName = String.format("insert_stream_%d_%d_%s_%d_onlinefs",
+              featureGroup.getFeatureStore().getProjectId(),
+              featureGroup.getId(),
+              featureGroup.getName(),
+              featureGroup.getVersion()
+      );
+    }
+    return queryName;
+  }
+
 }

java/spark/src/main/java/com/logicalclocks/hsfs/spark/engine/hudi/HudiEngine.java

@@ -264,7 +264,9 @@ private Map<String, String> setupHudiWriteOpts(FeatureGroupBase featureGroup, Hu
     hudiArgs.put(HUDI_HIVE_SYNC_DB, featureGroup.getFeatureStore().getName());
     hudiArgs.put(HIVE_AUTO_CREATE_DATABASE_OPT_KEY, HIVE_AUTO_CREATE_DATABASE_OPT_VAL);
     hudiArgs.put(HUDI_HIVE_SYNC_SUPPORT_TIMESTAMP, "true");
-    hudiArgs.put(HUDI_TABLE_OPERATION, operation.getValue());
+    if (operation != null) {
+      hudiArgs.put(HUDI_TABLE_OPERATION, operation.getValue());
+    }
     hudiArgs.putAll(HUDI_DEFAULT_PARALLELISM);
 
     // Overwrite with user provided options if any
@@ -307,8 +309,7 @@ private void createEmptyTable(SparkSession sparkSession, StreamFeatureGroup stre
       throws IOException, FeatureStoreException {
     Configuration configuration = sparkSession.sparkContext().hadoopConfiguration();
     Properties properties = new Properties();
-    properties.putAll(setupHudiWriteOpts(streamFeatureGroup,
-        HudiOperationType.BULK_INSERT, null));
+    properties.putAll(setupHudiWriteOpts(streamFeatureGroup, null, null));
     HoodieTableMetaClient.initTableAndGetMetaClient(configuration, streamFeatureGroup.getLocation(), properties);
   }
 

java/spark/src/test/java/com/logicalclocks/hsfs/spark/engine/TestSparkEngine.java

@@ -28,9 +28,7 @@
 import com.logicalclocks.hsfs.metadata.StorageConnectorApi;
 import com.logicalclocks.hsfs.spark.FeatureGroup;
 import com.logicalclocks.hsfs.spark.FeatureStore;
-import org.apache.spark.SparkContext;
 import org.apache.spark.sql.Dataset;
-import org.apache.spark.sql.SparkSession;
 import org.apache.spark.sql.types.DataTypes;
 import org.apache.spark.sql.types.StructField;
 import org.apache.spark.sql.types.StructType;
@@ -227,4 +225,32 @@ public void testGetKafkaConfigExternalClientInternalKafka() throws FeatureStoreE
         Mockito.verify(storageConnectorApi).getKafkaStorageConnector(Mockito.any(), externalArg.capture());
         Assertions.assertEquals(Boolean.FALSE, externalArg.getValue());
     }
+
+    @Test
+    public void testMakeQueryName() {
+        SparkEngine sparkEngine = SparkEngine.getInstance();
+        FeatureGroup featureGroup = new FeatureGroup();
+        Integer fgId = 1;
+        String fgName = "test_fg";
+        Integer fgVersion = 1;
+        Integer projectId = 99;
+        featureGroup.setId(fgId);
+        featureGroup.setName(fgName);
+        featureGroup.setVersion(fgVersion);
+        FeatureStore featureStore = (new FeatureStore());
+        featureStore.setProjectId(projectId);
+        featureGroup.setFeatureStore(featureStore);
+        String queryName = String.format("insert_stream_%d_%d_%s_%d_onlinefs",
+                featureGroup.getFeatureStore().getProjectId(),
+                featureGroup.getId(),
+                featureGroup.getName(),
+                featureGroup.getVersion()
+        );
+        // query name is null
+        Assertions.assertEquals(queryName, sparkEngine.makeQueryName(null, featureGroup));
+        // query name is empty
+        Assertions.assertEquals(queryName, sparkEngine.makeQueryName("", featureGroup));
+        // query name is not empty
+        Assertions.assertEquals("test_qn", sparkEngine.makeQueryName("test_qn", featureGroup));
+    }
 }

python/hsfs/client/auth.py

@@ -21,7 +21,7 @@ class BearerAuth(requests.auth.AuthBase):
     """Class to encapsulate a Bearer token."""
 
     def __init__(self, token):
-        self._token = token
+        self._token = token.strip()
 
     def __call__(self, r):
         r.headers["Authorization"] = "Bearer " + self._token
@@ -32,7 +32,7 @@ class ApiKeyAuth(requests.auth.AuthBase):
     """Class to encapsulate an API key."""
 
     def __init__(self, token):
-        self._token = token
+        self._token = token.strip()
 
     def __call__(self, r):
         r.headers["Authorization"] = "ApiKey " + self._token

python/hsfs/client/external.py

@@ -106,10 +106,12 @@ def __init__(
             with open(os.path.join(self._cert_folder, "material_passwd"), "w") as f:
                 f.write(str(credentials["password"]))
 
-        elif engine == "spark":
+        elif engine == "spark" or engine == "spark-no-metastore":
             _spark_session = SparkSession.builder.getOrCreate()
 
-            self.validate_spark_configuration(_spark_session)
+            self.validate_spark_configuration(
+                _spark_session, engine == "spark-no-metastore"
+            )
             with open(
                 _spark_session.conf.get("spark.hadoop.hops.ssl.keystores.passwd.name"),
                 "r",
@@ -123,7 +125,7 @@ def __init__(
                 "spark.hadoop.hops.ssl.keystore.name"
             )
 
-    def validate_spark_configuration(self, _spark_session):
+    def validate_spark_configuration(self, _spark_session, no_metastore):
         exception_text = "Spark is misconfigured for communication with Hopsworks, missing or invalid property: "
 
         configuration_dict = {
@@ -135,11 +137,13 @@ def validate_spark_configuration(self, _spark_session):
             "spark.hadoop.fs.hopsfs.impl": "io.hops.hopsfs.client.HopsFileSystem",
             "spark.hadoop.hops.ssl.keystores.passwd.name": None,
             "spark.hadoop.hops.ipc.server.ssl.enabled": "true",
-            "spark.sql.hive.metastore.jars": None,
             "spark.hadoop.client.rpc.ssl.enabled.protocol": "TLSv1.2",
-            "spark.hadoop.hive.metastore.uris": None,
         }
 
+        if not no_metastore:
+            configuration_dict["spark.hadoop.hive.metastore.uris"] = None
+            configuration_dict["spark.sql.hive.metastore.jars"] = None
+
         for key, value in configuration_dict.items():
             if not (
                 _spark_session.conf.get(key, "not_found") != "not_found"

python/hsfs/connection.py

@@ -20,6 +20,7 @@
 from requests.exceptions import ConnectionError
 
 from hsfs.decorators import connected, not_connected
+from hsfs.core.opensearch import OpenSearchClientSingleton
 from hsfs import engine, client, util, usage
 from hsfs.core import (
     feature_store_api,
@@ -216,6 +217,11 @@ def connect(self):
                 self._engine = "python"
             elif self._engine is not None and self._engine.lower() == "training":
                 self._engine = "training"
+            elif (
+                self._engine is not None
+                and self._engine.lower() == "spark-no-metastore"
+            ):
+                self._engine = "spark-no-metastore"
             else:
                 raise ConnectionError(
                     "Engine you are trying to initialize is unknown. "
@@ -271,6 +277,7 @@ def close(self):
             conn.close()
             ```
         """
+        OpenSearchClientSingleton().close()
         client.stop()
         self._feature_store_api = None
         engine.stop()