はじめに

オンプレFlink 2.0.1のKafka Source、Iceberg Sink、1分集計をManaged Service for Apache Flinkへ移す。

batch用のsyslog_icebergとauthlog_iceberg、streaming用の4 tableは、先行シリーズで構築したS3 Icebergをそのまま利用する。既存tableのlocationやmetadataを作り直さず、オンプレFlinkからManaged Service for Apache Flinkへwriterを切り替える。

操作端末では既存Iceberg環境とprofileを読み込む。

source /etc/iceberg/aws.env
export AWS_PROFILE=home-aws
aws sts get-caller-identity --profile "${AWS_PROFILE}"

Managed Service for Apache Flink本体はこのprofileを使用しない。S3/Glue/MSKへの権限はapplication roleへ付与する。

version差分

既存環境はFlink 2.0.1、Kafka SQL Connector 4.0.1-2.0、Iceberg Flink runtime 2.0/1.10.1である。Managed Serviceの対応runtimeは変化するため、作成画面と公式一覧を確認する。執筆時点の公式document historyにはFlink 2.2対応が掲載されている。

完全一致しないため、オンプレのSQL fileをそのままuploadするのではなく、選択したManaged Service runtime（本稿では2.2想定）でJava applicationをbuildし直す。

主な確認点は次のとおりである。

Table/SQL APIのbreaking change
Kafka connectorとMSK認証
Iceberg runtimeのFlink minor version一致
shaded JARとAWS runtime提供dependencyの重複
savepoint互換性とoperator UID

対象job

source	sink	用途
`syslog`	`syslog_events`	event stream
`authlog`	`authlog_events`	event stream
`syslog`	`syslog_host_1m`	1分×host集計
`authlog`	`authlog_host_1m`	1分×host集計

batch用syslog_iceberg/authlog_icebergはEMR Serverlessが担当し、Flinkと統合しない。

1. applicationを実装する

Java applicationでTableEnvironmentを作成し、既存SQLをresourceとして実行する。catalogをHiveCatalogからGlueCatalogへ、bootstrap serverをMSK private endpointへ変更する。

application起動時にruntime property groupのmsk.bootstrap.serversを読み、SQLへ渡す。JAR内へendpointを固定しない。

Map<String, Properties> groups = KinesisAnalyticsRuntime.getApplicationProperties();
String bootstrapServers = groups.get("ApplicationProperties")
    .getProperty("msk.bootstrap.servers");
String sourceDdl = Files.readString(
    Path.of(Main.class.getResource("/sql/syslog-source.sql").toURI()),
    StandardCharsets.UTF_8);
tableEnvironment.executeSql(
    sourceDdl.replace("${MSK_PRIVATE_BOOTSTRAP}", bootstrapServers));

CREATE TEMPORARY TABLE syslog_json_1m (
  ts STRING,
  src_host STRING,
  program STRING,
  message STRING,
  event_time AS TO_TIMESTAMP(REPLACE(SUBSTRING(ts, 1, 19), 'T', ' ')),
  WATERMARK FOR event_time AS event_time - INTERVAL '10' SECOND
) WITH (
  'connector'='kafka',
  'topic'='syslog',
  'properties.bootstrap.servers'='${MSK_PRIVATE_BOOTSTRAP}',
  'properties.group.id'='flink-syslog-host-1m-aws',
  'scan.startup.mode'='latest-offset',
  'format'='json',
  'json.ignore-parse-errors'='true'
);

集計SQLは既存どおり1分TUMBLE、10秒watermark、TRIM(src_host)、dt、hrを維持する。Iceberg側のprimary key/upsertは選択したIceberg/Flink組み合わせで検証し、同じminute/hostの重複を確認する。

2. Glue Iceberg catalogを設定する

application JARには選択runtime用Iceberg Flink runtimeとIceberg AWS bundleを含める。catalog warehouseは次に固定する。

s3://${ICEBERG_BUCKET}/warehouse

application roleへMSK connect/read、Glue read/write、warehouse prefix read/write、CloudWatch Logs権限を付与する。SASL/SCRAMならSecrets Manager/KMS権限も追加する。

export AWS_ADMIN_PROFILE=home-aws-admin
cat > /tmp/flink-trust.json <<'EOF'
{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"Service":"kinesisanalytics.amazonaws.com"},"Action":"sts:AssumeRole"}]}
EOF

FLINK_ROLE_ARN=$(aws iam create-role \
  --role-name home-log-managed-flink-role \
  --assume-role-policy-document file:///tmp/flink-trust.json \
  --query Role.Arn --output text --profile "${AWS_ADMIN_PROFILE}")

KMS_KEY_ARN=$(aws kms describe-key --region "${AWS_REGION}" \
  --key-id "${KMS_KEY_ID}" --query KeyMetadata.Arn --output text \
  --profile "${AWS_ADMIN_PROFILE}")

jq -n --arg lake "${ICEBERG_BUCKET}" --arg artifact "${ARTIFACT_BUCKET}" \
  --arg secret "${MSK_SECRET_ARN}" --arg key "${KMS_KEY_ARN}" \
  '{Version:"2012-10-17",Statement:[
    {Effect:"Allow",Action:["s3:ListBucket"],Resource:[("arn:aws:s3:::"+$lake),("arn:aws:s3:::"+$artifact)]},
    {Effect:"Allow",Action:["s3:GetObject","s3:PutObject","s3:DeleteObject"],Resource:[("arn:aws:s3:::"+$lake+"/warehouse/*"),("arn:aws:s3:::"+$artifact+"/flink/*")]},
    {Effect:"Allow",Action:["glue:GetDatabase","glue:GetDatabases","glue:GetTable","glue:GetTables","glue:CreateTable","glue:UpdateTable"],Resource:"*"},
    {Effect:"Allow",Action:["kafka:DescribeCluster","kafka:GetBootstrapBrokers"],Resource:"*"},
    {Effect:"Allow",Action:"secretsmanager:GetSecretValue",Resource:$secret},
    {Effect:"Allow",Action:"kms:Decrypt",Resource:$key},
    {Effect:"Allow",Action:["logs:DescribeLogGroups","logs:DescribeLogStreams","logs:PutLogEvents"],Resource:"*"},
    {Effect:"Allow",Action:["ec2:DescribeVpcs","ec2:DescribeSubnets","ec2:DescribeSecurityGroups","ec2:CreateNetworkInterface","ec2:DescribeNetworkInterfaces","ec2:DeleteNetworkInterface"],Resource:"*"}
  ]}' > /tmp/flink-policy.json

aws iam put-role-policy \
  --role-name home-log-managed-flink-role \
  --policy-name home-log-managed-flink-policy \
  --policy-document file:///tmp/flink-policy.json \
  --profile "${AWS_ADMIN_PROFILE}"

動作確認後、Glue、Kafka、LogsのResourceは実際のARNへ絞る。

3. networkを設定する

applicationをMSKへ到達できるprivate subnetとSecurity Groupへ関連付ける。MSK側inboundはapplication Security Groupから必要portのみ許可する。S3 Gateway Endpointと、NATを使わない場合に必要なAWS service endpointを確認する。

FLINK_SECURITY_GROUP_ID=$(aws ec2 create-security-group \
  --region "${AWS_REGION}" --vpc-id "${VPC_ID}" \
  --group-name home-log-flink-sg --description 'Managed Flink application' \
  --query GroupId --output text --profile "${AWS_ADMIN_PROFILE}")

aws ec2 authorize-security-group-ingress \
  --region "${AWS_REGION}" --group-id "${MSK_SECURITY_GROUP_ID}" \
  --protocol tcp --port 9096 --source-group "${FLINK_SECURITY_GROUP_ID}" \
  --profile "${AWS_ADMIN_PROFILE}"

aws ec2 create-vpc-endpoint \
  --region "${AWS_REGION}" --vpc-id "${VPC_ID}" \
  --service-name "com.amazonaws.${AWS_REGION}.s3" \
  --vpc-endpoint-type Gateway --route-table-ids "${ROUTE_TABLE_ID}" \
  --profile "${AWS_ADMIN_PROFILE}"

S3以外のGlue、CloudWatch Logs、Secrets Manager、KMSへNATなしで接続する場合は、それぞれのInterface VPC endpointも作成する。

4. JARを配置してapplicationを作る

aws s3 cp target/log-flink-app.jar "s3://$ARTIFACT_BUCKET/flink/log-flink-app.jar" \
  --profile home-aws-admin

AWS CLIでruntime、application role、JAR location、VPC、parallelism、CloudWatch logging、snapshotを設定する。FLINK_RUNTIMEは作成時点でManaged Serviceがサポートする値に合わせる。

export FLINK_RUNTIME=FLINK-1_20
MSK_PRIVATE_BOOTSTRAP=$(aws kafka get-bootstrap-brokers \
  --region "${AWS_REGION}" --cluster-arn "${MSK_CLUSTER_ARN}" \
  --query BootstrapBrokerStringSaslScram --output text \
  --profile "${AWS_ADMIN_PROFILE}")

aws logs create-log-group --region "${AWS_REGION}" \
  --log-group-name /aws/kinesis-analytics/home-log-flink \
  --profile "${AWS_ADMIN_PROFILE}" 2>/dev/null || true
aws logs create-log-stream --region "${AWS_REGION}" \
  --log-group-name /aws/kinesis-analytics/home-log-flink \
  --log-stream-name application \
  --profile "${AWS_ADMIN_PROFILE}" 2>/dev/null || true

LOG_STREAM_ARN="arn:aws:logs:${AWS_REGION}:$(aws sts get-caller-identity --query Account --output text --profile "${AWS_ADMIN_PROFILE}"):log-group:/aws/kinesis-analytics/home-log-flink:log-stream:application"

jq -n \
  --arg bucket_arn "arn:aws:s3:::${ARTIFACT_BUCKET}" \
  --arg bootstrap "${MSK_PRIVATE_BOOTSTRAP}" \
  --arg warehouse "s3://${ICEBERG_BUCKET}/warehouse" \
  --arg secret "${MSK_SECRET_ARN}" \
  --arg subnet1 "${SUBNET_ID_1}" --arg subnet2 "${SUBNET_ID_2}" \
  --arg subnet3 "${SUBNET_ID_3}" --arg sg "${FLINK_SECURITY_GROUP_ID}" \
  '{FlinkApplicationConfiguration:{
      CheckpointConfiguration:{ConfigurationType:"CUSTOM",CheckpointingEnabled:true,CheckpointInterval:300000,MinPauseBetweenCheckpoints:10000},
      MonitoringConfiguration:{ConfigurationType:"CUSTOM",MetricsLevel:"TASK",LogLevel:"INFO"},
      ParallelismConfiguration:{ConfigurationType:"CUSTOM",Parallelism:1,ParallelismPerKPU:1,AutoScalingEnabled:true}},
    EnvironmentProperties:{PropertyGroups:[{PropertyGroupId:"ApplicationProperties",PropertyMap:{
      "msk.bootstrap.servers":$bootstrap,"iceberg.warehouse":$warehouse,"msk.secret.arn":$secret}}]},
    ApplicationCodeConfiguration:{CodeContent:{S3ContentLocation:{BucketARN:$bucket_arn,FileKey:"flink/log-flink-app.jar"}},CodeContentType:"ZIPFILE"},
    ApplicationSnapshotConfiguration:{SnapshotsEnabled:true},
    VpcConfigurations:[{SubnetIds:[$subnet1,$subnet2,$subnet3],SecurityGroupIds:[$sg]}]}' \
  > /tmp/flink-application-config.json

aws kinesisanalyticsv2 create-application \
  --region "${AWS_REGION}" --application-name home-log-flink \
  --runtime-environment "${FLINK_RUNTIME}" \
  --service-execution-role "${FLINK_ROLE_ARN}" \
  --application-configuration file:///tmp/flink-application-config.json \
  --cloud-watch-logging-options "LogStreamARN=${LOG_STREAM_ARN}" \
  --application-mode STREAMING \
  --profile "${AWS_ADMIN_PROFILE}"

aws kinesisanalyticsv2 start-application \
  --region "${AWS_REGION}" --application-name home-log-flink \
  --profile "${AWS_ADMIN_PROFILE}"
aws kinesisanalyticsv2 describe-application \
  --region "${AWS_REGION}" --application-name home-log-flink \
  --query 'ApplicationDetail.ApplicationStatus' --output text \
  --profile "${AWS_ADMIN_PROFILE}"

checkpointとsavepoint

オンプレでは5分間隔、EXACTLY_ONCE、HDFS checkpoint/savepointを使用している。Managed Serviceではservice-managed checkpointとapplication snapshotを使う。S3 directoryをexecution.checkpointing.dirへ直接指定する運用とは異なる。

automatic snapshotを有効にする
update/停止前にsnapshotを作成する
operator UIDを明示し、state mappingを安定させる
checkpoint成功数、失敗数、durationをCloudWatchで監視する

Flink 2.0.1から異なるruntimeへの直接savepoint restoreは、connector/state serializer互換性を含めて保証せず、PoCでは新しいconsumer groupでlatest-offsetから並行起動する。過去offsetから必要なら開始offsetを固定して重複許容の再処理を行う。

5. 検証する

SELECT COUNT(*), MIN(window_start), MAX(window_start)
FROM logs.syslog_host_1m
WHERE dt = current_date;

SELECT host, SUM(cnt)
FROM logs.authlog_host_1m
WHERE dt = current_date
GROUP BY host;

Managed Service dashboard、CloudWatch、MSK consumer lag、Glue metadata、Athena queryを確認する。applicationをsnapshot付きで停止・再開し、重複/欠損とcheckpoint復旧を確認してから旧Flink jobを止める。

参考資料

関連記事

実機検証について

本稿は移植設計であり未検証である。Flink 2.2、Iceberg、Kafka connectorの組み合わせはbuild時にdependency treeとintegration testで確定する。