Databricks Auto Loaderのファイル通知モードを試してみた

# 背景・目的
前回、[Databricks Auto Loaderを試してみた](https://qiita.com/zumax/items/4aa8352025641c381de2)では、Auto Loaderについて知識の整理や、簡単なハンズオンを試してみました。

今回は、Auto Loaderのファイル通知モードを試してみます。

# まとめ
ファイル通知モードで登場するAWSリソースは下記の通り。（だと思います。）

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/9bd62e2b-c556-8c5b-1d6a-f90b03de9126.png)


# 概要
ファイル通知モードの簡単な整理については、[こちら](https://qiita.com/zumax/items/4aa8352025641c381de2#%E3%83%95%E3%82%A1%E3%82%A4%E3%83%AB%E9%80%9A%E7%9F%A5%E3%83%A2%E3%83%BC%E3%83%89)にまとめました。

# 実践
[こちら](https://docs.databricks.com/ja/ingestion/auto-loader/file-notification-mode.html#required-permissions-for-configuring-file-notification-for-aws-s3)を元に試します

## AWS コンソールを使用してインスタンスプロファイルを作成する
1. AWSにサインインし、IAMに移動します
1. IAMで「ロールを作成」をクリックします
1. 「AWSのサービス」＞ユースケース「EC2」を選択し、「次へ」をクリックします
1. ポリシーは選択せずに、「次へ」をクリックします
1. ロール名を入力し、「ロールを作成」をクリックします
1. 作成したロールを選択します
1. 許可を追加＞インラインポリシーを作成をクリックします
    <img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/4d0e6396-60e3-2bc9-f0dd-bb1956e831c4.png" width=300 />
1. 下記を貼り付け、「次へ」を選択します
    ```
    {
      "Version": "2012-10-17",
      "Statement": [
        {
          "Effect": "Allow",
          "Action": [
            "s3:ListBucket"
          ],
         "Resource": [
            "arn:aws:s3:::<s3-bucket-name>"
          ]
        },
        {
          "Effect": "Allow",
          "Action": [
            "s3:PutObject",
            "s3:GetObject",
            "s3:DeleteObject",
            "s3:PutObjectAcl"
          ],
          "Resource": [
             "arn:aws:s3:::<s3-bucket-name>/*"
          ]
        }
      ]
    }
    ```
1. ポリシー名を入力し、「ポリシーの作成」をクリックします

## バケットポリシーを作成する
1. 下記のバケットポリシーをS3にアタッチします
    ```
    {
      "Version": "2012-10-17",
      "Statement": [
        {
          "Sid": "Example permissions",
          "Effect": "Allow",
          "Principal": {
            "AWS": "arn:aws:iam::<aws-account-id-databricks>:role/<iam-role-for-s3-access>"
          },
          "Action": [
            "s3:GetBucketLocation",
            "s3:ListBucket"
          ],
          "Resource": "arn:aws:s3:::<s3-bucket-name>"
        },
        {
          "Effect": "Allow",
          "Principal": {
            "AWS": "arn:aws:iam::<aws-account-id-databricks>:role/<iam-role-for-s3-access>"
          },
          "Action": [
            "s3:PutObject",
            "s3:GetObject",
            "s3:DeleteObject",
            "s3:PutObjectAcl"
          ],
          "Resource": "arn:aws:s3:::<s3-bucket-name>/*"
        }
      ]
    }
    ```
## Databricks デプロイを作成した IAMロールを見つける
1. アカウントコンソールにログインします
1. ワークスペースを選択し、CredentialsボックスのロールARNの末尾にあるロール名をコピーします

## EC2 ポリシーに S3 IAMロールを追加する
1. AWSのIAMに移動します
1. IAMロールから、上記のDatabricksをデプロイで作成したIAMロールを選択します
1. 下記をStatement配列の末尾に追加します
    ```
        {
          "Effect": "Allow",
          "Action": "iam:PassRole",
          "Resource": "arn:aws:iam::<aws-account-id-databricks>:role/<iam-role-for-s3-access>"
        }
    ```
1. 変更を保存します

## インスタンスプロファイルを Databricks に追加する
1. ワークスペース管理者としてワークスペースにサインインします
1. 画面右上の個人メニューから「Settings」をクリックします
1. 「Security」をクリックします
1. Instance profilesの「Manage」をクリックします
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/d0ba7602-0255-ccb6-bc82-b1cd1aa0506b.png)
1. 「Add instance profile」をクリックします
1. 作成したいIAMロールのInstance profile ARNを貼り付け、「Add」をクリックします
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/8ef55a81-f6f3-03be-fa81-d998d8c40a5e.png)



## IAMポリシーの作成とアタッチ
1. 下記のIAMポリシーを作成します
    ```
    {
      "Version": "2012-10-17",
      "Statement": [
        {
          "Sid": "DatabricksAutoLoaderSetup",
          "Effect": "Allow",
          "Action": [
            "s3:GetBucketNotification",
            "s3:PutBucketNotification",
            "sns:ListSubscriptionsByTopic",
            "sns:GetTopicAttributes",
            "sns:SetTopicAttributes",
            "sns:CreateTopic",
            "sns:TagResource",
            "sns:Publish",
            "sns:Subscribe",
            "sqs:CreateQueue",
            "sqs:DeleteMessage",
            "sqs:ReceiveMessage",
            "sqs:SendMessage",
            "sqs:GetQueueUrl",
            "sqs:GetQueueAttributes",
            "sqs:SetQueueAttributes",
            "sqs:TagQueue",
            "sqs:ChangeMessageVisibility"
          ],
          "Resource": [
            "arn:aws:s3:::<bucket-name>",
            "arn:aws:sqs:<region>:<account-number>:databricks-auto-ingest-*",
            "arn:aws:sns:<region>:<account-number>:databricks-auto-ingest-*"
          ]
        },
        {
          "Sid": "DatabricksAutoLoaderList",
          "Effect": "Allow",
          "Action": [
            "sqs:ListQueues",
            "sqs:ListQueueTags",
            "sns:ListTopics"
          ],
          "Resource": "*"
        },
        {
          "Sid": "DatabricksAutoLoaderTeardown",
          "Effect": "Allow",
          "Action": [
            "sns:Unsubscribe",
            "sns:DeleteTopic",
            "sqs:DeleteQueue"
          ],
          "Resource": [
            "arn:aws:sqs:<region>:<account-number>:databricks-auto-ingest-*",
            "arn:aws:sns:<region>:<account-number>:databricks-auto-ingest-*"
          ]
        }
      ]
    }
    ```
1. IAMロールにアタッチします

## クラスタにInstance profileをアタッチ
1. Computeをクリックします
1. 対象のClusterを選択し、「Edit」をクリックします
1. Instance profileで作成したインスタンスプロファイルを選択し、保存します

## ファイルの用意
1. ファイルを用意します
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/bf9be61f-91cf-a5fc-797f-560adfb7206b.png)


## コード
1. Notebookを作成
1. 下記のコードを貼り付けます。`option("cloudFiles.useNotifications","true")`を追加しています
    ```
    # Import functions
    from pyspark.sql.functions import col, current_timestamp
        
    # Define variables used in code below
    file_path = "s3://XXXXX/test2"
    table_name = f"autoloader_test_file_notice"
    checkpoint_path = f"s3://XXXXX/etl_quickstart/autoloader_test_file_notice"
        
    # Clear out data from previous demo execution
    spark.sql(f"DROP TABLE IF EXISTS {table_name}")
    dbutils.fs.rm(checkpoint_path, True)
        
    # Configure Auto Loader to ingest JSON data to a Delta table
    (spark.readStream
          .format("cloudFiles")
          .option("cloudFiles.format", "json")
          .option("cloudFiles.schemaLocation", checkpoint_path)
          .option("inferSchema","true")
          .option("cloudFiles.useNotifications","true")
          .load(file_path)
          .select("*", col("_metadata.file_path").alias("source_file"), current_timestamp().alias("processing_time"))
          .writeStream
          .option("checkpointLocation", checkpoint_path)
          .trigger(availableNow=True)
          .toTable(table_name))        
    ```
1. 実行します
1. 下記のコードを実行します
    ```
    df = spark.read.table(table_name)
    display(df)
    ```
1. 表示されました
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/77fbac20-7ae7-86c0-4cb4-f3a4bb01f8c5.png)

## AWSリソースの確認
上記の実行により、AWSリソースが作成されているので確認します

### S3の確認
1. S3のトップページに移動します
1. 対象のバケットを選択し、「プロパティ」をクリックします
1. イベント通知では、送信先にSNSが指定されています
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/695db4f1-89db-3541-b8ed-c0378db5c194.png)

### SNSの確認
1. SNSのトップページに移動します
1. トピックを選択します
1. トピックが作成されているので、クリックします
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/d8d78d54-92bf-3f35-b7b3-6a4a465dc433.png)
1. サブスクリプションをクリックします
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/7451cc87-21ee-350d-b778-5c3977c04290.png)
1. エンドポイントに、SQSが指定されています
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/86d38704-59bb-8f5e-468c-9979cd319285.png)

### SQSの確認
1. SQSトップページに移動します
1. 作成されていました。クリックします
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/1d173e51-a019-991a-d4d5-0f19cb1f6322.png)

# 考察
今回、ファイル通知モードで、Auto Loaderを構成しました。ディレクトリリストモードと比較してスケーラビリティが高い構成とのことで、大量データを扱う場合は良いのかもしれませんね。
しかし、ドキュメントにも記載があるように、ストレージサービス（S3）のファイル通知の仕組みに依存しており、100%を保証していないので、バックフィルが必須とのことでした。次回はバックフィルについて調べてみたいと思います。


# 参考
https://qiita.com/zumax/items/4aa8352025641c381de2

https://docs.databricks.com/ja/ingestion/auto-loader/file-notification-mode.html

https://docs.databricks.com/ja/connect/storage/tutorial-s3-instance-profile.html