KARTEのイベントデータをAPI経由で取得して別のプロジェクトにイベントを転送する

<html><head></head><body>こんにちは、Customer Engineerのyazyです！最近グッピーに子供が生まれ、育休を取るかどうか悩んでいます！
さて、KARTEではジョブフローを使ってファイルを取り込むことはできますが、通常機能では取り込んだファイルを削除することはできません。しかし、Craft Functionsを使うことで取り込み後のファイルを削除することが可能になります。
そこで今回は、Craft Functionsを使ってジョブフローの実行後にS3上のファイルの削除を実現する方法について紹介します。
<h1 id="アウトプットイメージ">アウトプットイメージ</h1>
構成図は次の通り。
<img src="https://ik.imagekit.io/plaid/a8f1cedb-057c-43c4-8744-d51db03e06de/%E5%85%A8%E4%BD%93%E6%A7%8B%E6%88%90%E5%9B%B33.png" alt="全体構成図3.png">
ジョブフローからSQLの情報を利用してS3上のファイルを削除します。
<h1 id="注意点">注意点</h1>
前提として、次の点に注意をしてください。
<ul>
<li>S3へのアクセスはアクセスキー/シークレットアクセスキーを利用していることを前提としています</li>
<li>作成されるIAMにファイルを削除するための権限が付与されている前提としています</li>
<li>CraftのDatahub Jobflow Connectorを利用しますので、Growthプラン以上で実現可能です</li>
</ul>
<h1 id="設定手順">設定手順</h1>
設定の手順は次の通りです。
<ol>
<li>AWS IAM で取得した取得したIAMのアクセスキー/シークレットアクセスキーをCraft Secret Manager に登録する</li>
<li>Craft Functionsのファンクションを作成する</li>
<li>SQLクエリを作成する</li>
<li>ジョブフローを作成する</li>
</ol>
順番に見てみましょう。
<h2 id="1. AWS IAM で取得した取得したIAMのアクセスキー/シークレットアクセスキーをCraft Secret Manager に登録する">1. AWS IAM で取得した取得したIAMのアクセスキー/シークレットアクセスキーをCraft Secret Manager に登録する</h2>
作成したアクセスキー/シークレットアクセスキーをCraft Functionsから安全に利用するために、Craft Secret Manager に登録します。
<img src="https://ik.imagekit.io/plaid/de246508-2c91-40ac-a04a-c214af6542c7/create_slack_token_secret.png" alt="create_slack_token_secret.png">
<h3 id="下記の手順で、アクセスキーとシークレットアクセスキーをそれぞれ登録してください。">下記の手順で、アクセスキーとシークレットアクセスキーをそれぞれ登録してください。</h3>
<ul>
<li>[Craft &gt; シークレット] からシークレットを新規作成</li>
<li>[名前] に、わかりやすい名前を入力</li>
<li>メモしておいたAWS IAMのアクセスキー/シークレットアクセスキーの値を [シークレットの値] に入力し、作成</li>
<li>シークレット名をメモしておく</li>
</ul>
<h2 id="2. Craft Functionsのファンクションを作成する">2. Craft Functionsのファンクションを作成する</h2>
Craft Functionsのファンクションを作成します。作成手順は次の通りです。
<img src="https://ik.imagekit.io/plaid/f54845aa-6834-475f-ad9d-dfaf4b0ef4c5/create_craft_functions.png" alt="create_craft_functions.png">
<ul>
<li>[Craft &gt; ファンクション &gt; 新規作成 &gt; テンプレートから作成] を選択</li>
<li>「 KARTE Datahubジョブフローの実行時に指定したS3バケット上のファイルを削除する 」というテンプレートを検索し[取得]ボタンをクリック</li>
<li>[反映] ボタンをクリック</li>
<li>[変数] タブで次の変数の値を設定する
<ul>
<li><code>ACCESS_KEY_SECRET</code>
<ul>
<li>アクセスキーのシークレット名を入力</li>
</ul>
</li>
<li><code>SECRET_ACCESS_KEY_SECRET</code>
<ul>
<li>シークレットアクセスキーのシークレット名を入力</li>
</ul>
</li>
<li>[設定 &gt; ファンクションの有効化] にチェックを付けて [保存]</li>
</ul>
</li>
</ul>
なお、テンプレートのソースコードはGitHubで公開しています。 
<a href="https://github.com/plaidev/craft-codes/tree/main/craft-functions/s3-import-cleanup" target="_blank" rel="noreferrer">craft-codes/craft-functions/s3-import-cleanup at main · plaidev/craft-codes</a>
<h2 id="3. SQLクエリを作成する">3. SQLクエリを作成する</h2>
SQLクエリを作成します。作成手順は次の通りです。ここでは削除対象のS3上のファイルを指定します。
<img src="https://ik.imagekit.io/plaid/1ca11f93-3efb-4e47-9a43-18599356b230/SQL%E7%B5%90%E6%9E%9C%E7%94%BB%E5%83%8F.png" alt="SQL結果画像.png">
<ul>
<li>[Datahub &gt; クエリ &gt; 作成 &gt; クエリを作成] を選択</li>
<li>[名前] に、わかりやすい名前を入力</li>
<li>カラム名は以下になるように作成します
<ul>
<li>region
<ul>
<li>該当のS3バケットが存在するリージョンを記述</li>
</ul>
</li>
<li>bucket_name
<ul>
<li>該当のS3バケット名を記述</li>
</ul>
</li>
<li>file_path
<ul>
<li>該当のS3バケットに存在するファイルのパスを記述</li>
<li>こちら削除対象ファイルとなります</li>
</ul>
</li>
</ul>
</li>
</ul>
2つのファイルを削除する場合は以下のように記述します。
<pre><code class="hljs">WITH target_file_info AS
(
SELECT 'ap-northeast-1' AS region, 'karte-data-bucket' AS bucket_name, 'karte-065dae9b89d32a5b556cd3cd2b8df357/test.csv' AS file_path
UNION ALL
SELECT 'ap-northeast-1' AS region, 'karte-data-bucket' AS bucket_name, 'karte-065dae9b89d32a5b556cd3cd2b8df357/test2.csv' AS file_path
)

SELECT * from target_file_info
</code></pre>
次のようなクエリ結果になれば大丈夫です。
<img src="https://ik.imagekit.io/plaid/c1b32667-2d84-42c7-be14-ff6b1f28b682/SQL%E7%B5%90%E6%9E%9C%E7%94%BB%E5%83%8F.png" alt="SQL結果画像.png">
<h2 id="4. ジョブフローを作成する">4. ジョブフローを作成する</h2>
ジョブフローを作成します。作成手順は次の通りです。
<img src="https://ik.imagekit.io/plaid/c85edb9c-08bc-4cd0-9e5b-04b6013728fb/%E3%82%B7%E3%82%99%E3%83%A7%E3%83%95%E3%82%99%E3%83%95%E3%83%AD%E3%83%BC%E7%94%BB%E5%83%8F.png" alt="ジョブフロー画像.png">
<ul>
<li>[Datahub &gt; ジョブフロー &gt; 作成] を選択</li>
<li>[ジョブの追加 &gt; クエリ実行結果のエクスポート &gt; 次へ] を選択</li>
<li>[クエリ選択] で、手順3で作成したSQLクエリ名を選択</li>
<li>[外部サービス &gt; 接続先] で、Craftを選択</li>
<li>[個別設定 &gt; function名]に、手順1で作成したFunction名を入力</li>
<li>[行ごとにqueueをpublishする]にチェックを入れる</li>
</ul>
設定が終了したら、ジョブフローを実行してみましょう！SQL上で指定したファイルが削除されます。
実際には、S3バケットからデータをインポートするジョブフローの中で、インポート完了後にこのファイル削除ジョブが実行されるよう設定することを想定しています。
<h1 id="おわりに">おわりに</h1>
今回はCraft FunctionsでS3上のファイルをDatahubの実行後に削除する方法について紹介しました。
S3バケット上のファイル容量を削減したい場合や、Datahubでのインポートジョブ実行後に対象ファイルを削除しなければいけないシステム要件がある場合にご活用いただけます。ぜひ皆さんも試してみてください！
<hr>
プレイドでは、世の中にインパクトを与える新しいソリューションを一緒につくる仲間を募集しています！ぜひ気軽にお問い合わせください。
↓ パートナー企業になりたい
<ul>
<li><a href="https://ecosystem.plaid.co.jp/" target="_blank" rel="noreferrer">CustomerData-Driven Solution を一緒に創りませんか？ | PLAID Ecosystem</a></li>
</ul>
↓ PLAIDで働きたい
<ul>
<li><a href="https://recruit.plaid.co.jp/requisitions/general-customer-engineer" target="_blank" rel="noreferrer">Customer Engineer | 募集ポジション | 株式会社プレイド</a></li>
<li><a href="https://recruit.plaid.co.jp/requisitions/product-engineer" target="_blank" rel="noreferrer">Product Engineer | 募集ポジション | 株式会社プレイド</a></li>
</ul>
</body></html>

KARTE Datahubジョブフロー実行時に、指定したS3バケット上のファイルを削除します。削除対象のS3バケットのリージョン、バケット名、削除対象ファイルパスはDatahubクエリで指定します。