Azure Storage で構築した Snowflake Open Catalog の外部テーブルに INSERT する際のエラーに対する暫定対応方法

## 概要

Snowflake Open Catalogのチュートリアルである`Getting started with Snowflake Open Catalog`のユースケース1をAzure Storageで実施した際、INSERT処理がエラーとなった際の対応方法を共有します。OSS の Polaris  Catalog のドキュメントにて、下記のように記述されているため、Spark では動作しないようです。なお、本記事での検証はGoogle ColabのSpark環境を利用しています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9b889130-2608-42f6-8c01-7dc42b00e415.png)

> Apache Spark can’t use credential vending

出所：[Overview | Apache Polaris](https://polaris.apache.org/in-dev/unreleased/overview/#credential-vending)

## Snowflake Open Catalog に関する基本的な情報

本記事の作業内容は、Sparkを使ってSnowflake Open Catalogを通じ、Azure Storage（今回はS3 bucketではなくAzure Storage）にデータを書き込む処理です。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9bc2ce40-2fb7-4a37-9e26-b23de70d8cd8.png)

出所：[Getting started with Snowflake Open Catalog | Snowflake Documentation](https://other-docs.snowflake.com/en/opencatalog/tutorials/open-catalog-gs)

公式ドキュメントでは以下のコードが提示されていますが、このままではエラーが発生しました。

```python
import pyspark
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('iceberg_lab') \
    .config('spark.jars.packages', 'org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.4.1,org.apache.iceberg:iceberg-azure-bundle:1.5.2') \
    .config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
    .config('spark.sql.defaultCatalog', 'opencatalog') \
    .config('spark.sql.catalog.opencatalog', 'org.apache.iceberg.spark.SparkCatalog') \
    .config('spark.sql.catalog.opencatalog.type', 'rest') \
    .config('spark.sql.catalog.opencatalog.uri','https://ab12345.snowflakecomputing.com/polaris/api/catalog') \
    .config('spark.sql.catalog.opencatalog.header.X-Iceberg-Access-Delegation','vended-credentials') \
    .config('spark.sql.catalog.opencatalog.credential','000000000000000000000000000=:1111111111111111111111111111111111111111111=') \
    .config('spark.sql.catalog.opencatalog.warehouse','Catalog1') \
    .config('spark.sql.catalog.opencatalog.scope','PRINCIPAL_ROLE:data_engineer') \
    .getOrCreate()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/af033beb-8eda-4bac-bfbb-56ec88b76068.png)


出所：[サービス接続の登録 | Snowflake Documentation](https://other-docs.snowflake.com/ja/opencatalog/register-service-connection#example-4-register-a-single-service-connection-azure)

公式ドキュメントでは、BlobとADLSの両方が指定可能となっていますが、ADLSでの動作方法については十分に確認できていません。また、BlobのURIについて、公式ドキュメントでは「abfss」と記載されていますが、実際は「wasbs」が正しい表記となります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/aadcf084-b78a-49be-9f60-d92b8e8b816d.png)

出所：[Create a catalog | Snowflake Documentation](https://other-docs.snowflake.com/en/opencatalog/create-catalog#create-a-catalog-using-azure-storage)

## 事前準備

### Google Colab 上でノートブックを作成して Spark が動作することを確認

```bash
!pyspark --version
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ccdc50d3-e651-4c4b-ae5c-0235bc1471da.png)

### Blob エンドポイントの Snowflake Open Catalog カタログを作成

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ed8afec6-d154-4ba9-98c5-0195ffa185a2.png)

## 正常に動作した方法

### Blob エンドポイントを登録したカタログに対する INSERT 処理

```python
# Snowflake Open Catalog に関する情報をセット
open_catalog_account_identifier = "lampvrn-test123"
client_id = "8bq6hVXHOxV5ZKJQfb="
client_secret = "52/mD7sDxUQWjgZgxGCx37lO2x02fV="
catalog_name = "manabian_10"
principal_role_name = "spark_all"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/41a8a0f2-d083-4e38-a313-e2e960a35a4b.png)


```python
# Azure Storage に関る情報をセット
azure_storage_account_name = "snowflakeiceberg0123456"
azure_storage_account_key = "kKRn6nruDHEhUF0AQzM1PFlXtXs9V0BJQSrn6Z8GvJOsb0JHflV9Fn=="
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f1681be4-1e1b-4be1-b3a8-d76db27c2b22.png)


```python
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('iceberg_lab') \
    .config('spark.jars.packages',
            'org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.4.1,'
            'org.apache.iceberg:iceberg-azure-bundle:1.5.2,'
            'org.apache.hadoop:hadoop-azure:3.4.1') \
    .config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
    .config('spark.sql.defaultCatalog', 'opencatalog') \
    .config('spark.sql.catalog.opencatalog', 'org.apache.iceberg.spark.SparkCatalog') \
    .config('spark.sql.catalog.opencatalog.type', 'rest') \
    .config('spark.sql.catalog.opencatalog.header.X-Iceberg-Access-Delegation','vended-credentials') \
    .config('spark.sql.catalog.opencatalog.uri',f'https://{open_catalog_account_identifier}.snowflakecomputing.com/polaris/api/catalog') \
    .config('spark.sql.catalog.opencatalog.credential',f'{client_id}:{client_secret}') \
    .config('spark.sql.catalog.opencatalog.warehouse',catalog_name) \
    .config('spark.sql.catalog.opencatalog.scope',f'PRINCIPAL_ROLE:{principal_role_name}') \
    .config(f'spark.hadoop.fs.azure.account.key.{azure_storage_account_name}.blob.core.windows.net', azure_storage_account_key) \
    .getOrCreate()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3c894b94-60b1-4280-bbd0-c0557ec54a52.png)


```python
spark.sql("create namespace if not exists spark_demo")
spark.sql("create or replace table spark_demo.test_table (col1 int) using iceberg")
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/76f32717-71dc-4ed7-a368-c27c51562b45.png)

```python
# insert a record in the table
spark.sql("insert into spark_demo.test_table values (1)");

spark.table("spark_demo.test_table").show()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/475fb3b2-dee4-4383-8ede-e26ca6da4743.png)

## エラーとなった方法

### ドキュメントのまま実行する方法

```python
# Snowflake Open Catalog に関する情報をセット
open_catalog_account_identifier = "lampvrn-test123"
client_id = "8bq6hVXHOxV5ZKJQfb="
client_secret = "52/mD7sDxUQWjgZgxGCx37lO2x02fV="
catalog_name = "manabian_10"
principal_role_name = "spark_all"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/41a8a0f2-d083-4e38-a313-e2e960a35a4b.png)

```python
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('iceberg_lab') \
    .config('spark.jars.packages', 'org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.4.1,com.azure:azure-storage-blob:12.14.2') \
    .config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
    .config('spark.sql.defaultCatalog', 'opencatalog') \
    .config('spark.sql.catalog.opencatalog', 'org.apache.iceberg.spark.SparkCatalog') \
    .config('spark.sql.catalog.opencatalog.type', 'rest') \
    .config('spark.sql.catalog.opencatalog.header.X-Iceberg-Access-Delegation','vended-credentials') \
    .config('spark.sql.catalog.opencatalog.uri',f'https://{open_catalog_account_identifier}.snowflakecomputing.com/polaris/api/catalog') \
    .config('spark.sql.catalog.opencatalog.credential',f'{client_id}:{client_secret}') \
    .config('spark.sql.catalog.opencatalog.warehouse',catalog_name) \
    .config('spark.sql.catalog.opencatalog.scope',f'PRINCIPAL_ROLE:{principal_role_name}') \
    .getOrCreate()
```

```python
#insert a record in the table
spark.sql("insert into spark_demo.test_table values (1)");
```

> Py4JJavaError: An error occurred while calling o47.sql.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 0.0 failed 1 times, most recent failure: Lost task 0.0 in stage 0.0 (TID 0) (3754e32b0da4 executor driver): java.lang.RuntimeException: java.lang.ClassNotFoundException: Class org.apache.hadoop.fs.azure.NativeAzureFileSystem$Secure not found

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/233bf71f-2e5d-4763-93ba-e77293a77fd8.png)


### org.apache.hadoop:hadoop-azure を追加


```python
# Snowflake Open Catalog に関する情報をセット
open_catalog_account_identifier = "lampvrn-test123"
client_id = "8bq6hVXHOxV5ZKJQfb="
client_secret = "52/mD7sDxUQWjgZgxGCx37lO2x02fV="
catalog_name = "manabian_10"
principal_role_name = "spark_all"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/41a8a0f2-d083-4e38-a313-e2e960a35a4b.png)

```python
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('iceberg_lab') \
    .config('spark.jars.packages',
            'org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.4.1,'
            'org.apache.iceberg:iceberg-azure-bundle:1.5.2,'
            'org.apache.hadoop:hadoop-azure:3.4.1') \
    .config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
    .config('spark.sql.defaultCatalog', 'opencatalog') \
    .config('spark.sql.catalog.opencatalog', 'org.apache.iceberg.spark.SparkCatalog') \
    .config('spark.sql.catalog.opencatalog.type', 'rest') \
    .config('spark.sql.catalog.opencatalog.header.X-Iceberg-Access-Delegation','vended-credentials') \
    .config('spark.sql.catalog.opencatalog.uri',f'https://{open_catalog_account_identifier}.snowflakecomputing.com/polaris/api/catalog') \
    .config('spark.sql.catalog.opencatalog.credential',f'{client_id}:{client_secret}') \
    .config('spark.sql.catalog.opencatalog.warehouse',catalog_name) \
    .config('spark.sql.catalog.opencatalog.scope',f'PRINCIPAL_ROLE:{principal_role_name}') \
    .getOrCreate()
```

```python
# insert a record in the table
spark.sql("insert into spark_demo.test_table values (1)");
```

> Py4JJavaError: An error occurred while calling o47.sql.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 1.0 failed 1 times, most recent failure: Lost task 0.0 in stage 1.0 (TID 1) (2a8859124b60 executor driver): org.apache.iceberg.exceptions.RuntimeIOException: Failed to get file system for path: wasbs://snowflake@snowflake.blob.core.windows.net/blob_test/spark_demo/test_table/data/00000-1-0261d99b-77be-4b63-b670-1d71b31f93c2-00001.parquet

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e33c65da-80b1-48fa-a10e-edf927ae5786.png)

### ADLS エンドポイントを登録したカタログに対する INSERT 処理


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a8604777-c0df-40ea-a8fd-d9d20b164372.png)


```python
# Snowflake Open Catalog に関する情報をセット
open_catalog_account_identifier = "lampvrn-test123"
client_id = "8bq6hVXHOxV5ZKJQfb="
client_secret = "52/mD7sDxUQWjgZgxGCx37lO2x02fV="
catalog_name = "manabian_10__abfs"
principal_role_name = "spark_all"
```

```python
# Azure Storage に関る情報をセット
azure_storage_account_name = "snowflakeiceberg0123456"
azure_storage_account_key = "kKRn6nruDHEhUF0AQzM1PFlXtXs9V0BJQSrn6Z8GvJOsb0JHflV9Fn=="
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/97574218-3b42-4541-85d1-c56c809e111a.png)


```python
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('iceberg_lab') \
    .config('spark.jars.packages',
            'org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.4.1,'
            'org.apache.iceberg:iceberg-azure-bundle:1.5.2,'
            'org.apache.hadoop:hadoop-azure:3.4.1') \
    .config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
    .config('spark.sql.defaultCatalog', 'opencatalog') \
    .config('spark.sql.catalog.opencatalog', 'org.apache.iceberg.spark.SparkCatalog') \
    .config('spark.sql.catalog.opencatalog.type', 'rest') \
    .config('spark.sql.catalog.opencatalog.header.X-Iceberg-Access-Delegation','vended-credentials') \
    .config('spark.sql.catalog.opencatalog.uri',f'https://{open_catalog_account_identifier}.snowflakecomputing.com/polaris/api/catalog') \
    .config('spark.sql.catalog.opencatalog.credential',f'{client_id}:{client_secret}') \
    .config('spark.sql.catalog.opencatalog.warehouse',catalog_name) \
    .config('spark.sql.catalog.opencatalog.scope',f'PRINCIPAL_ROLE:{principal_role_name}') \
    .config(f'spark.hadoop.fs.azure.account.key.{azure_storage_account_name}.dfs.core.windows.net', azure_storage_account_key) \
    .getOrCreate()
```

```python
spark.sql("create namespace if not exists spark_demo")
spark.sql("create or replace table spark_demo.test_table12 (col1 int) using iceberg")
```

> Py4JJavaError: An error occurred while calling o51.sql.
: org.apache.iceberg.exceptions.BadRequestException: Malformed request: Cannot create table on external catalogs.

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c54ede76-e517-4804-8d80-1da8a04643d2.png)