Examples

Spark query for data across clouds

Afon deploys Polaris as a REST catalog server.

Spark connects to Polaris, which vends short-lived cloud credentials so Spark can read and write Iceberg tables stored in S3, GCS, or Azure Blob Storage — all from a single Spark session.

Generate PySpark code for Jupyter:

afon-cli.sh generate-py-spark-context

This prints ready-to-run PySpark code that:

1. Creates a Spark session connected to the Spark Connect server
2. Writes test data to AWS, Azure and GCP catalogs
3. Joins data across all three clouds in a single query

The generated code looks like:

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("test").remote("sc://spark-connector-0-driver-svc:15002").getOrCreate()

df_test = spark.createDataFrame([["test", 1]], ['name', 'num'])

# AWS
aws_namespace = "aws.acceptancetests"
spark.sql(f"CREATE NAMESPACE IF NOT EXISTS {aws_namespace}")
df_test.writeTo(f"{aws_namespace}.smoketesttable").create()

# Azure
az_namespace = "azure.acceptancetests"
spark.sql(f"CREATE NAMESPACE IF NOT EXISTS {az_namespace}")
df_test.writeTo(f"{az_namespace}.smoketesttable").append()

# GCP
gcp_namespace = "gcp.acceptancetests"
spark.sql(f"CREATE NAMESPACE IF NOT EXISTS {gcp_namespace}")
df_test.writeTo(f"{gcp_namespace}.smoketesttable").create()

# Cross-cloud join
df_gcp = spark.read.table(f"{gcp_namespace}.smoketesttable")
df_azure = spark.read.table(f"{az_namespace}.smoketesttable")
df_aws = spark.read.table(f"{aws_namespace}.smoketesttable")

df_joined = (
    df_gcp
    .join(df_azure, on="name", how="inner")
    .join(df_aws, on="name", how="inner")
)
df_joined.show()

Write and read an Iceberg table via Spark AWS

To test writing and reading to and from an Iceberg table via Spark on aws for catalog azure_external:

./afon.sh -a exec -ec ops.spark.SparkTestWriter -ea "aws azure_external test_namespace test_table"

This creates a test namespace and table, and writes three rows (“apple”, “banana”, “cherry”).

./afon.sh -a exec -ec ops.spark.SparkTestReader -ea "aws azure_external test_namespace test_table"

This reads the test table created.

NOTE: Both SparkTestWriter and SparkTestReader accept as optional args the credentials (username and password) of the user to authenticate as. If missing, the user named by property USER_ID is used, with its password from users.csv.

Snowflake reads GCP table via Polaris

The SQL generated by manage-stack run aws-snowflake reading-gcp. The token comes from KeyCloak, as Polaris returns HTTP 501 from its own token endpoint, and the objects are named after the catalog, eg iceberg_azure_id for azure. Snowflake cancels a statement it has not finished after 10 minutes (SnowflakeConnectionFactory.QUERY_TIMEOUT_SECONDS):

CREATE OR REPLACE CATALOG INTEGRATION iceberg_gcp_id
  CATALOG_SOURCE = ICEBERG_REST
  TABLE_FORMAT = ICEBERG
  REST_CONFIG = (
    CATALOG_URI = 'https://emryspolaris.click/api/catalog/'
    CATALOG_NAME = 'gcp'
    ACCESS_DELEGATION_MODE = VENDED_CREDENTIALS
  )
  REST_AUTHENTICATION = (
    TYPE = OAUTH
    OAUTH_TOKEN_URI = 'https://emrysoauth.click/realms/polaris/protocol/openid-connect/token'
    OAUTH_CLIENT_ID = '<polaris_id>'
    OAUTH_CLIENT_SECRET = '<polaris_secret>'
    OAUTH_ALLOWED_SCOPES = ('openid')
  )
  ENABLED = TRUE;

USE DATABASE SNOWFLAKE_LEARNING_DB;

CREATE OR REPLACE ICEBERG TABLE my_gcp_table_id
  CATALOG = 'iceberg_gcp_id'
  CATALOG_TABLE_NAME = 'test_table'
  CATALOG_NAMESPACE = 'namespaceph'
  REPLACE_INVALID_CHARACTERS = TRUE;

SELECT COUNT(*) FROM my_gcp_table_id;

Snowflake writes Iceberg table to Azure

CREATE OR REPLACE STORAGE INTEGRATION azure_int
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = AZURE
  ENABLED = TRUE
  STORAGE_ALLOWED_LOCATIONS = ('azure://afonstorageaccount.blob.core.windows.net/afonstorage')
  AZURE_TENANT_ID='YOUR_TENANT_ID';

CREATE OR REPLACE EXTERNAL VOLUME my_external_volume
  STORAGE_LOCATIONS = (
    (
      NAME = 'azure_loc'
      STORAGE_PROVIDER = 'AZURE'
      STORAGE_BASE_URL = 'azure://afonstorageaccount.blob.core.windows.net/afonstorage/acceptancetests/'
      AZURE_TENANT_ID='YOUR_TENANT_ID'
    )
  );

USE DATABASE SNOWFLAKE_LEARNING_DB;

CREATE OR REPLACE ICEBERG TABLE my_azure_iceberg_table (
    value STRING
)
  EXTERNAL_VOLUME = 'MY_EXTERNAL_VOLUME'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_project_folder/';

INSERT INTO my_azure_iceberg_table (value)
VALUES ('Alice'), ('Bob');

After writing, you can register the table in AWS Polaris so Spark can also read it.