do more stuff nested namespaces for iceberg, make everything tables and make replacement logic where none could be found by default. Why is there no sane defaults for spark???

Author: reinthal

nessie-stack/jaffle-shop-classic-spark/dbt_project.yml

@@ -1,5 +1,6 @@
 name: 'jaffle_shop'
 
+
 config-version: 2
 version: '0.1'
 
@@ -21,7 +22,18 @@ require-dbt-version: [">=1.0.0", "<2.0.0"]
 
 models:
   jaffle_shop:
+      +on_table_exists: replace
       +table_format: iceberg
       materialized: table
       staging:
-        materialized: view
+        +schema: stg
+        materialized: table
+
+seeds:
+  jaffle_shop:
+    +quote_columns: false
+    +pre_hook: "DROP TABLE IF EXISTS {{ this }}"
+    +schema: raw
+    +file_format: iceberg
+    +materialized: seed
+    

nessie-stack/jaffle-shop-classic-spark/macros/generate_schema_name.sql

@@ -0,0 +1,14 @@
+{% macro generate_schema_name(custom_schema_name, node) -%}
+
+    {%- set default_schema = target.schema -%}
+    {%- if custom_schema_name is none -%}
+
+        {{ default_schema }}
+
+    {%- else -%}
+
+        {{ default_schema }}.{{ custom_schema_name | trim }}
+
+    {%- endif -%}
+
+{%- endmacro %}

nessie-stack/jaffle-shop-classic-spark/models/employees2.sql

@@ -1,3 +1,10 @@
+{{
+  config(
+    materialized='table',
+    file_format='iceberg'
+  )
+}}
+
 with source as (
 
     {#-

nessie-stack/jaffle-shop-classic-spark/models/sources.yml

@@ -9,7 +9,7 @@
      "name": "stderr",
      "output_type": "stream",
      "text": [
-      "25/03/21 22:59:45 WARN SparkSession: Using an existing Spark session; only runtime SQL configurations will take effect.\n"
+      "25/03/22 00:13:18 WARN SparkSession: Using an existing Spark session; only runtime SQL configurations will take effect.\n"
      ]
     }
    ],
@@ -28,13 +28,14 @@
      "data": {
       "text/plain": [
        "[('spark.sql.catalog.nessie', 'org.apache.iceberg.spark.SparkCatalog'),\n",
+       " ('spark.app.submitTime', '1742601784570'),\n",
        " ('spark.sql.catalog.nessie.uri', 'http://nessie:19120/api/v1'),\n",
        " ('spark.hadoop.fs.s3a.path.style.access', 'true'),\n",
-       " ('spark.app.id', 'local-1742597984957'),\n",
        " ('spark.sql.catalog.nessie.ref', 'main'),\n",
+       " ('spark.app.startTime', '1742601784818'),\n",
+       " ('spark.driver.port', '39441'),\n",
        " ('spark.sql.warehouse.dir',\n",
        "  'file:/home/iceberg/notebooks/notebooks/spark-warehouse'),\n",
-       " ('spark.driver.port', '40237'),\n",
        " ('spark.hadoop.fs.s3a.access.key', 'minioadmin'),\n",
        " ('spark.sql.catalog.nessie.s3.path-style-access', 'true'),\n",
        " ('spark.serializer.objectStreamReset', '100'),\n",
@@ -49,31 +50,30 @@
        "  'file:///root/.ivy2/jars/org.apache.iceberg_iceberg-spark-runtime-3.5_2.12-1.8.1.jar,file:///root/.ivy2/jars/org.projectnessie.nessie-integrations_nessie-spark-extensions-3.5_2.12-0.103.2.jar'),\n",
        " ('spark.driver.extraJavaOptions',\n",
        "  '-Djava.net.preferIPv6Addresses=false -XX:+IgnoreUnrecognizedVMOptions --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.lang.invoke=ALL-UNNAMED --add-opens=java.base/java.lang.reflect=ALL-UNNAMED --add-opens=java.base/java.io=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.util.concurrent=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.base/jdk.internal.ref=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/sun.nio.cs=ALL-UNNAMED --add-opens=java.base/sun.security.action=ALL-UNNAMED --add-opens=java.base/sun.util.calendar=ALL-UNNAMED --add-opens=java.security.jgss/sun.security.krb5=ALL-UNNAMED -Djdk.reflect.useDirectMethodHandle=false'),\n",
-       " ('spark.driver.host', '41c5f4f73c86'),\n",
+       " ('spark.app.id', 'local-1742601785395'),\n",
        " ('spark.sql.catalog.nessie.s3.endpoint', 'http://minio:9000'),\n",
        " ('spark.repl.local.jars',\n",
        "  'file:///root/.ivy2/jars/org.apache.iceberg_iceberg-spark-runtime-3.5_2.12-1.8.1.jar,file:///root/.ivy2/jars/org.projectnessie.nessie-integrations_nessie-spark-extensions-3.5_2.12-0.103.2.jar'),\n",
        " ('spark.executor.id', 'driver'),\n",
-       " ('spark.app.submitTime', '1742597984280'),\n",
+       " ('spark.driver.host', '876c347e2cee'),\n",
        " ('spark.app.name', 'PySparkShell'),\n",
        " ('spark.hadoop.fs.s3a.impl', 'org.apache.hadoop.fs.s3a.S3AFileSystem'),\n",
        " ('spark.sql.catalogImplementation', 'hive'),\n",
        " ('spark.sql.catalog.nessie.io-impl', 'org.apache.iceberg.aws.s3.S3FileIO'),\n",
        " ('spark.submit.pyFiles',\n",
        "  '/root/.ivy2/jars/org.apache.iceberg_iceberg-spark-runtime-3.5_2.12-1.8.1.jar,/root/.ivy2/jars/org.projectnessie.nessie-integrations_nessie-spark-extensions-3.5_2.12-0.103.2.jar'),\n",
        " ('spark.rdd.compress', 'True'),\n",
-       " ('spark.app.initial.jar.urls',\n",
-       "  'spark://41c5f4f73c86:40237/jars/org.projectnessie.nessie-integrations_nessie-spark-extensions-3.5_2.12-0.103.2.jar,spark://41c5f4f73c86:40237/jars/org.apache.iceberg_iceberg-spark-runtime-3.5_2.12-1.8.1.jar'),\n",
        " ('spark.executor.extraJavaOptions',\n",
        "  '-Djava.net.preferIPv6Addresses=false -XX:+IgnoreUnrecognizedVMOptions --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.lang.invoke=ALL-UNNAMED --add-opens=java.base/java.lang.reflect=ALL-UNNAMED --add-opens=java.base/java.io=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.util.concurrent=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.base/jdk.internal.ref=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/sun.nio.cs=ALL-UNNAMED --add-opens=java.base/sun.security.action=ALL-UNNAMED --add-opens=java.base/sun.util.calendar=ALL-UNNAMED --add-opens=java.security.jgss/sun.security.krb5=ALL-UNNAMED -Djdk.reflect.useDirectMethodHandle=false'),\n",
        " ('spark.sql.catalog.nessie.warehouse', 's3a://warehouse'),\n",
        " ('spark.jars',\n",
        "  'file:///root/.ivy2/jars/org.apache.iceberg_iceberg-spark-runtime-3.5_2.12-1.8.1.jar,file:///root/.ivy2/jars/org.projectnessie.nessie-integrations_nessie-spark-extensions-3.5_2.12-0.103.2.jar'),\n",
-       " ('spark.app.startTime', '1742597984489'),\n",
        " ('spark.hadoop.fs.s3a.endpoint', 'http://minio:9000'),\n",
        " ('spark.files',\n",
        "  'file:///root/.ivy2/jars/org.apache.iceberg_iceberg-spark-runtime-3.5_2.12-1.8.1.jar,file:///root/.ivy2/jars/org.projectnessie.nessie-integrations_nessie-spark-extensions-3.5_2.12-0.103.2.jar'),\n",
        " ('spark.sql.catalog.nessie.type', 'nessie'),\n",
+       " ('spark.app.initial.jar.urls',\n",
+       "  'spark://876c347e2cee:39441/jars/org.apache.iceberg_iceberg-spark-runtime-3.5_2.12-1.8.1.jar,spark://876c347e2cee:39441/jars/org.projectnessie.nessie-integrations_nessie-spark-extensions-3.5_2.12-0.103.2.jar'),\n",
        " ('spark.ui.showConsoleProgress', 'true')]"
       ]
      },
@@ -88,16 +88,27 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 4,
+   "execution_count": 3,
    "metadata": {},
    "outputs": [
+    {
+     "name": "stderr",
+     "output_type": "stream",
+     "text": [
+      "25/03/22 00:13:22 WARN HiveConf: HiveConf of name hive.stats.jdbc.timeout does not exist\n",
+      "25/03/22 00:13:22 WARN HiveConf: HiveConf of name hive.stats.retries.wait does not exist\n",
+      "25/03/22 00:13:22 WARN ObjectStore: Version information not found in metastore. hive.metastore.schema.verification is not enabled so recording the schema version 2.3.0\n",
+      "25/03/22 00:13:22 WARN ObjectStore: setMetaStoreSchemaVersion called but recording version is disabled: version = 2.3.0, comment = Set by MetaStore [email protected]\n",
+      "25/03/22 00:13:23 WARN ObjectStore: Failed to get database global_temp, returning NoSuchObjectException\n"
+     ]
+    },
     {
      "data": {
       "text/plain": [
        "DataFrame[]"
       ]
      },
-     "execution_count": 4,
+     "execution_count": 3,
      "metadata": {},
      "output_type": "execute_result"
     }
@@ -108,7 +119,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 11,
+   "execution_count": 4,
    "metadata": {},
    "outputs": [
     {
@@ -118,7 +129,7 @@
      "traceback": [
       "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
       "\u001b[0;31mAnalysisException\u001b[0m                         Traceback (most recent call last)",
-      "Cell \u001b[0;32mIn[11], line 1\u001b[0m\n\u001b[0;32m----> 1\u001b[0m \u001b[43mspark\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43msql\u001b[49m\u001b[43m(\u001b[49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[38;5;124;43mcreate namespace default\u001b[39;49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[43m)\u001b[49m\n",
+      "Cell \u001b[0;32mIn[4], line 1\u001b[0m\n\u001b[0;32m----> 1\u001b[0m \u001b[43mspark\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43msql\u001b[49m\u001b[43m(\u001b[49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[38;5;124;43mcreate namespace default\u001b[39;49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[43m)\u001b[49m\n",
       "File \u001b[0;32m/opt/spark/python/pyspark/sql/session.py:1631\u001b[0m, in \u001b[0;36mSparkSession.sql\u001b[0;34m(self, sqlQuery, args, **kwargs)\u001b[0m\n\u001b[1;32m   1627\u001b[0m         \u001b[38;5;28;01massert\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_jvm \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m\n\u001b[1;32m   1628\u001b[0m         litArgs \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_jvm\u001b[38;5;241m.\u001b[39mPythonUtils\u001b[38;5;241m.\u001b[39mtoArray(\n\u001b[1;32m   1629\u001b[0m             [_to_java_column(lit(v)) \u001b[38;5;28;01mfor\u001b[39;00m v \u001b[38;5;129;01min\u001b[39;00m (args \u001b[38;5;129;01mor\u001b[39;00m [])]\n\u001b[1;32m   1630\u001b[0m         )\n\u001b[0;32m-> 1631\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m DataFrame(\u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_jsparkSession\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43msql\u001b[49m\u001b[43m(\u001b[49m\u001b[43msqlQuery\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mlitArgs\u001b[49m\u001b[43m)\u001b[49m, \u001b[38;5;28mself\u001b[39m)\n\u001b[1;32m   1632\u001b[0m \u001b[38;5;28;01mfinally\u001b[39;00m:\n\u001b[1;32m   1633\u001b[0m     \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28mlen\u001b[39m(kwargs) \u001b[38;5;241m>\u001b[39m \u001b[38;5;241m0\u001b[39m:\n",
       "File \u001b[0;32m/opt/spark/python/lib/py4j-0.10.9.7-src.zip/py4j/java_gateway.py:1322\u001b[0m, in \u001b[0;36mJavaMember.__call__\u001b[0;34m(self, *args)\u001b[0m\n\u001b[1;32m   1316\u001b[0m command \u001b[38;5;241m=\u001b[39m proto\u001b[38;5;241m.\u001b[39mCALL_COMMAND_NAME \u001b[38;5;241m+\u001b[39m\\\n\u001b[1;32m   1317\u001b[0m     \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mcommand_header \u001b[38;5;241m+\u001b[39m\\\n\u001b[1;32m   1318\u001b[0m     args_command \u001b[38;5;241m+\u001b[39m\\\n\u001b[1;32m   1319\u001b[0m     proto\u001b[38;5;241m.\u001b[39mEND_COMMAND_PART\n\u001b[1;32m   1321\u001b[0m answer \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mgateway_client\u001b[38;5;241m.\u001b[39msend_command(command)\n\u001b[0;32m-> 1322\u001b[0m return_value \u001b[38;5;241m=\u001b[39m \u001b[43mget_return_value\u001b[49m\u001b[43m(\u001b[49m\n\u001b[1;32m   1323\u001b[0m \u001b[43m    \u001b[49m\u001b[43manswer\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mgateway_client\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mtarget_id\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mname\u001b[49m\u001b[43m)\u001b[49m\n\u001b[1;32m   1325\u001b[0m \u001b[38;5;28;01mfor\u001b[39;00m temp_arg \u001b[38;5;129;01min\u001b[39;00m temp_args:\n\u001b[1;32m   1326\u001b[0m     \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28mhasattr\u001b[39m(temp_arg, \u001b[38;5;124m\"\u001b[39m\u001b[38;5;124m_detach\u001b[39m\u001b[38;5;124m\"\u001b[39m):\n",
       "File \u001b[0;32m/opt/spark/python/pyspark/errors/exceptions/captured.py:185\u001b[0m, in \u001b[0;36mcapture_sql_exception.<locals>.deco\u001b[0;34m(*a, **kw)\u001b[0m\n\u001b[1;32m    181\u001b[0m converted \u001b[38;5;241m=\u001b[39m convert_exception(e\u001b[38;5;241m.\u001b[39mjava_exception)\n\u001b[1;32m    182\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m \u001b[38;5;28misinstance\u001b[39m(converted, UnknownException):\n\u001b[1;32m    183\u001b[0m     \u001b[38;5;66;03m# Hide where the exception came from that shows a non-Pythonic\u001b[39;00m\n\u001b[1;32m    184\u001b[0m     \u001b[38;5;66;03m# JVM exception message.\u001b[39;00m\n\u001b[0;32m--> 185\u001b[0m     \u001b[38;5;28;01mraise\u001b[39;00m converted \u001b[38;5;28;01mfrom\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[38;5;28;01mNone\u001b[39;00m\n\u001b[1;32m    186\u001b[0m \u001b[38;5;28;01melse\u001b[39;00m:\n\u001b[1;32m    187\u001b[0m     \u001b[38;5;28;01mraise\u001b[39;00m\n",
@@ -153,7 +164,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 13,
+   "execution_count": 5,
    "metadata": {},
    "outputs": [
     {