「JAWS-UG DE&I AWS re:Invent 2024 recap サーバーレスハンズオン」でサポーターしてきました

2025/01/26

JAWS-UG DE&I開催の「AWS re:Invent 2024 recap サーバーレスハンズオン」にハンズオンお手伝いとして参加しました。

いつもすてきな会場を提供いただいているアールスリーインスティテュートさんのgusuku Ashibinaa OSAKA！
ありがとうございます！

お土産コーナーにはAWS re:Invent2024で配布されていたステッカーやお菓子などなどがありました。

JAWS-UG大阪やDE&Iももう10周年とのことです。

Serverless Data Processing with AWS Lambda and Apache Kafka解説

配車ならぬ配ユニコーンサービスみたいなのをMSK, Lambda, AppSyncで作成するハンズオン。
最初にAWS下川さんから各サービスの解説がありました。
ありがたい！

型指定されたスキーマにクエリーを投げるのがGraphQL。
クライアント側が取得したいフィールドをリクエストできるなどが嬉しいポイント。
ワークショップではSubscriptionとMutationを使ってDynamoDBテーブルのアイテムに変化があればイベントを実行する。

KafkaはREST APIでアクセス可能で、イベントのバスとしても、データストアとしても使えるOSSのストリーミング。
MSK(Amazon Managed Streaming for Apache Kafka)はKafkaのマネージドサービス。
Serverlessで容量性能の見積もりなく動的にスケールできる。

KafkaのパーティションはKinesisのシャード。
パーティションキーがハッシュされてパーティションに分かれて。パーティションの中で順序性が保たれる。
KafkaのオフセットがKinesisのシーケンス番号。

Lambda Event Source MappingsでKafka、Kinesis、SQS、DynamoDBストリームなどを設定すると、ポーラーは実装しなくてもLambdaサービスがやってくれる。

Serverless Data Processing with AWS Lambda and Apache Kafkaハンズオン

環境にリソースがすでに作成されていました。

VPC

MSKとVS Code Server用のEC2インスタンスが起動するVPCです。

MSKは複数のプライベートサブネットで起動していてNATゲートウェイへのルートがあり、ここからインターネットへアウトバウンド通信が可能です。
MSKに設定されているセキュリティグループのインバウンドはKafkaへの通信のために9098ポートへすべての送信元からの通信が可能で、アウトバウンドはすべてのポートと送信先への通信が可能です。

VS Code Serverはパブリックサブネットで起動していて、セキュリティグループのインバウンドは80ポートですべての送信元から通信可能で、アウトバウンドはすべてのポートと送信先への通信が可能です。
フロントエンドのデプロイやMSKへのプロデューサーコマンドはVS Code Serverから実行します。

MSK(Amazon Managed Streaming for Apache Kafka)

MSK Serverlessのクラスターが作成されていました。

AppSync

AppSyncでGraphQL APIが作成されていました。

Lambda関数

すでに作成されていたLambda関数に次のコードと環境変数API_CHANNELの修正をしました。

import json
import os
import requests
import base64

API_CHANNEL = os.environ['API_CHANNEL']
API_URL = os.environ['API_URL']
API_KEY = os.environ['API_KEY']

session = requests.Session()

def lambda_handler(event, context):

   for name, messages in event["records"].items():

      for message in messages:
            query = 'mutation PublishData {{ publish(data: "{}", name: "{}") {{ data name }} }}'
            decoded_message = base64.b64decode(message["value"]).decode('utf-8')
            payload = {"msg": decoded_message.replace('"',"'")}

            merged_query = query.format(json.dumps(payload).replace('"','\\"'), API_CHANNEL)
            response = session.request( url=API_URL, method='POST', headers={'x-api-key': API_KEY}  ,json={'query': merged_query})
            print(response.json())

   return {
      'statusCode': 200,
      'name': 'message relay',
      'body': json.dumps({"msg":"unicorn telemetry updates!"})
   }

import json

import os

import requests

import base64

API_CHANNEL = os.environ['API_CHANNEL']

API_URL = os.environ['API_URL']

API_KEY = os.environ['API_KEY']

session = requests.Session()

def lambda_handler(event, context):

for name, messages in event["records"].items():

for message in messages:

query = 'mutation PublishData {{ publish(data: "{}", name: "{}") {{ data name }} }}'

decoded_message = base64.b64decode(message["value"]).decode('utf-8')

payload = {"msg": decoded_message.replace('"',"'")}

merged_query = query.format(json.dumps(payload).replace('"','\\"'), API_CHANNEL)

response = session.request( url=API_URL, method='POST', headers={'x-api-key': API_KEY} ,json={'query': merged_query})

print(response.json())

return {

'statusCode': 200,

'name': 'message relay',

'body': json.dumps({"msg":"unicorn telemetry updates!"})

}

MSKから受け取ったrecord.itemsからvalueをGraphQL APIにPOSTされています。

Lambda関数自体はVPC設定ではありません。
イベントトリガーにMSKを設定すると、ENIが追加されます。
ENIはMSKクラスターが配置されているサブネットに作成されたので3つのAZのサブネットに作成されました。

複数のトリガーを設定するとその数だけENIも作成されます。

少し気になったのは試しにイベントトリガーを削除してもENIは削除されずに、ENIを手動で削除やデタッチしようとするとアタッチされているLambdaがあるのでエラーとなりますが、そのLambdaが内部のサービスのようで操作できません。
CLIとかで操作したらできるかもですが、今回はコンソールでしか試していません。