AWS Step FunctionsでJSON LinesデータをS3 BucketにPut Object/Get Objectできるのか確認してみた

結論：Get Objectはできました。

#AWS Step Functions

#AWS

若槻龍太

2022.06.25

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

こんにちは、CX事業本部 IoT事業部の若槻です。

以前のエントリでStep FunctionsからのPut Object/Get Objectを行いましたが、その際にPutおよびGetを試したデータはJson Objectでした。

//Json Object
[
  {
    "key1": "val1",
    "key2": "val2",
    "key3": "val3"
  },
  {
    "key1": "val10",
    "key2": "val20",
    "key3": "val30"
  },
  {
    "key1": "val100",
    "key2": "val200",
    "key3": "val300"
  }
]

しかしBucketに格納されるJsonデータは、各レコードが改行記号（\n）で区切られたJSON Linesの方が Amazon Athenaなどによる分析用途には適しています。

//JSON Lines
{"key1":"val1","key2":"val2","key3":"val3"}
{"key1":"val10","key2":"val20","key2":"val30"}
{"key1":"val100","key2":"val200","key2":"val300"}

そこで今回は、AWS Step FunctionsでJSON LinesデータをS3 BucketにPut Object/Get Objectできるのか確認してみました。

確認してみた

Put Object（できなかった）

AWS CDK v2（TypeScript）で次のようなCDKスタックを作成します。

import { Construct } from 'constructs';
import {
  aws_s3,
  aws_stepfunctions,
  aws_stepfunctions_tasks,
  RemovalPolicy,
  Stack,
  StackProps,
} from 'aws-cdk-lib';

export class ProcessStack extends Stack {
  constructor(scope: Construct, id: string, props: StackProps) {
    super(scope, id, props);

    // S3 Bucket
    const s3Bucket = new aws_s3.Bucket(this, 'dataBucket', {
      removalPolicy: RemovalPolicy.DESTROY,
    });

    // Json ObjectをJSON Linesに変換
    const convertJsonObjectToJsonLinesTask =
      new aws_stepfunctions_tasks.EvaluateExpression(
        this,
        'convertJsonObjectToJsonLinesTask',
        {
          expression: aws_stepfunctions.JsonPath.format(
            '{}.map( x => JSON.stringify(x) ).join("\\n")',
            aws_stepfunctions.JsonPath.stringAt('$.object'),
          ),
          resultPath: '$.convertJsonObjectToJsonLinesTask',
        },
      );

    // Put Object
    const putObjectTask = new aws_stepfunctions_tasks.CallAwsService(
      this,
      'putObjectTask',
      {
        service: 's3',
        action: 'putObject',
        parameters: {
          Body: aws_stepfunctions.JsonPath.stringAt(
            '$.convertJsonObjectToJsonLinesTask',
          ),
          Bucket: s3Bucket.bucketName,
          Key: 'data/putObject',
        },
        iamResources: [`${s3Bucket.bucketArn}/*`],
        iamAction: 's3:PutObject',
        resultPath: aws_stepfunctions.DISCARD,
      },
    );

    // State Machine
    new aws_stepfunctions.StateMachine(this, 'stateMachine', {
      stateMachineName: 'stateMachine',
      definition: convertJsonObjectToJsonLinesTask.next(putObjectTask),
    });
  }
}

上記をCDK Deployしてスタックをデプロイします。

次の入力を指定してステートマシンを実行します。

{
  "object": [
    {
      "key1": "val1",
      "key2": "val2",
      "key3": "val3"
    },
    {
      "key1": "val10",
      "key2": "val20",
      "key3": "val30"
    },
    {
      "key1": "val100",
      "key2": "val200",
      "key3": "val300"
    }
  ]
}

実行が成功しました。putObjectTaskのParameterを見るとエスケープされたJSON Linesデータが文字列としてBodyに指定されています。

そしてBucket内に作成されたObjectを開いてみると、JSON Linesではなくエスケープされた文字列のまま書き込まれてしまっています。

JSON LinesではないのでS3 Selectでクエリを掛けても上手くパースできません。

またPut Object時のContentTypeをapplication/jsonに指定するなど試してみましたが、だめでした。

Get Object（できた）

AWS CDK v2（TypeScript）で次のようなCDKスタックを作成します。

import { Construct } from 'constructs';
import {
  aws_s3,
  aws_stepfunctions,
  aws_stepfunctions_tasks,
  RemovalPolicy,
  Stack,
  StackProps,
} from 'aws-cdk-lib';

export class ProcessStack extends Stack {
  constructor(scope: Construct, id: string, props: StackProps) {
    super(scope, id, props);

    // S3 Bucket
    const s3Bucket = new aws_s3.Bucket(this, 'dataBucket', {
      removalPolicy: RemovalPolicy.DESTROY,
    });

    // Get Object
    const getObjectTask = new aws_stepfunctions_tasks.CallAwsService(
      this,
      'getObjectTask',
      {
        service: 's3',
        action: 'getObject',
        parameters: {
          Bucket: s3Bucket.bucketName,
          Key: 'data/jsonlObject',
        },
        iamResources: [`${s3Bucket.bucketArn}/*`],
        iamAction: 's3:GetObject',
        resultPath: '$.getObjectTask',
      },
    );

    // JSON LinesをJson Objectに変換
    const convertJsonLinesToJsonObjectTask =
      new aws_stepfunctions_tasks.EvaluateExpression(
        this,
        'convertJsonLinesToJsonObjectTask',
        {
          expression: aws_stepfunctions.JsonPath.format(
            '`{}`.split("\\n").map( d => JSON.parse(d) )',
            aws_stepfunctions.JsonPath.stringAt('$.getObjectTask.Body'),
          ),
          resultPath: '$.convertJsonLinesToJsonObjectTask',
        },
      );

    // State Machine
    new aws_stepfunctions.StateMachine(this, 'stateMachine', {
      stateMachineName: 'stateMachine',
      definition: getObjectTask.next(convertJsonLinesToJsonObjectTask),
    });
  }
}

次のJSON Linesのデータを書き込んだオブジェクトをS3 Bucketに格納します。