ItemReader(맵)

Distributed Map 상태는 워크플로의 이전 단계에서 전달된 JSON 입력을 허용할 수 있습니다.

입력은 JSON 배열, JSON 객체 또는 JSON 객체의 노드 내 배열일 수 있습니다.

Step Functions는 배열의 요소 또는 JSON 객체의 키-값 페어에 대해 직접 반복합니다.

JSON 객체의 배열이 포함된 특정 노드를 선택하려면 ItemsPath(맵, JSONPath만 해당)를 사용하거나 JSONata 상태의 Items 필드에 JSONata 표현식을 사용할 수 있습니다.

개별 항목을 처리하기 위해 Distributed Map 상태는 각 항목에 대해 하위 워크플로 실행을 시작합니다. 다음 탭에서는 Map 상태에 전달된 입력과 하위 워크플로 실행에 대한 해당 입력의 예제를 보여줍니다.

Distributed Map 상태는 Amazon S3 버킷에 저장된 객체를 반복할 수 있습니다. 워크플로 실행이 Map 상태에 도달하면 Step Functions는 ListObjectsV2 API 작업을 호출하여 Amazon S3 객체 메타데이터의 배열을 반환합니다. 이 배열에서 각 항목에는 버킷에 저장된 실제 데이터에 대한 ETag 및 Key와 같은 데이터가 포함됩니다.

배열의 개별 항목을 처리하기 위해 Distributed Map 상태는 하위 워크플로 실행을 시작합니다. 예를 들어 Amazon S3 버킷에 이미지 100개가 포함되어 있다고 가정해보겠습니다. 그런 다음 ListObjectsV2 API 작업을 호출한 후 반환된 배열에는 100개의 메타데이터 항목이 포함됩니다. 그런 다음 Distributed Map 상태는 하위 워크플로 실행 100개를 시작하여 각 항목을 처리합니다.

중첩된 워크플로 없이 데이터 객체를 직접 처리하려면 LOAD_AND_FLATTEN 변환 옵션을 선택하여 항목을 직접 처리할 수 있습니다.

다음 탭에서는 이 데이터 세트의 하위 워크플로 실행에 전달된 ItemReader 필드 구문과 입력의 예제를 보여줍니다.

분산 맵의 입력 소스로 S3 ListObjectsV2에 대한 향상된 지원을 통해 상태 시스템은 Amazon S3 버킷에서 여러 데이터 객체를 직접 읽고 처리할 수 있으므로 중첩 맵이 메타데이터를 처리할 필요가 없습니다!

LOAD_AND_FLATTEN 옵션을 사용하면 상태 시스템이 다음을 수행합니다.

변환 옵션을 사용하면 메타데이터를 처리하는 데 더 이상 중첩 분산 맵이 필요하지 않습니다. LOAD_AND_FLATTEN 옵션을 사용하면 확장성이 향상되고 활성 맵 실행 수가 줄어들며 여러 객체를 동시에 처리합니다.

다음 구성은에 대한 설정을 보여줍니다. ItemReader

"ItemReader": {
   "Resource": "arn:aws:states:::s3:listObjectsV2",
   "ReaderConfig": {
      "InputType": "JSON",
      "Transformation": "LOAD_AND_FLATTEN"
   },
   "Arguments": {
      "Bucket": "S3_BUCKET_NAME",
      "Prefix": "S3_BUCKET_PREFIX"
   }
}

Distributed Map 상태는 Amazon S3 버킷에 저장된 JSON 파일을 데이터 세트로 허용할 수 있습니다. JSON 파일에는 배열 또는 JSON 객체가 포함되어야 합니다.

워크플로 실행이 Map 상태에 도달하면 Step Functions는 GetObject API 작업을 간접적으로 호출하여 지정된 JSON 파일을 가져옵니다.

JSON 파일에 중첩된 객체 구조가 포함된 경우를 사용하여 데이터 세트가 있는 특정 노드를 선택할 수 있습니다ItemsPointer. 예를 들어 다음 구성은 인벤토리에 있는 추천 제품의 중첩된 목록을 추출합니다.

"ItemReader": {
   "Resource": "arn:aws:states:::s3:getObject",
   "ReaderConfig": {
      "InputType": "JSON",
      "ItemsPointer": "/inventory/products/featured"
   },
   "Arguments": {
      "Bucket": "amzn-s3-demo-bucket",
      "Key": "nested-data-file.json"
   }
}

그러면 Map 상태가 배열의 각 항목을 반복하고 항목마다 하위 워크플로 실행을 시작합니다. 예를 들어 JSON 파일에 배열 항목 1,000개가 포함된 경우 Map 상태는 하위 워크플로 실행 1,000개를 시작합니다.

다음 탭에서는 이 데이터 세트의 하위 워크플로 실행에 전달된 ItemReader 필드 구문과 입력의 예제를 보여줍니다.

이 예제의 경우 factcheck.json이라는 JSON 파일이 있다고 가정해보겠습니다. 이 파일은 Amazon S3 버킷의 jsonDataset 접두사 내에 저장되었습니다. 다음은 JSON 데이터 세트의 예제입니다.

[
  {
    "verdict": "true",
    "statement_date": "6/11/2008",
    "statement_source": "speech"
  },
  {
    "verdict": "false",
    "statement_date": "6/7/2022",
    "statement_source": "television"
  },
  {
    "verdict": "mostly-true",
    "statement_date": "5/18/2016",
    "statement_source": "news"
  },
  ...
]

Distributed Map 상태는 Amazon S3 버킷에 데이터 세트로 저장된 JSON Lines 파일을 수락할 수 있습니다.

다음 탭에서는 이 데이터 세트의 하위 워크플로 실행에 전달된 ItemReader 필드 구문과 입력의 예제를 보여줍니다.

이 예제에서는 이름이 인 JSON Lines 파일이 있다고 가정합니다factcheck.jsonl. 이 파일은 Amazon S3 버킷의 jsonlDataset 접두사 내에 저장되었습니다. 다음은 파일 콘텐츠의 예입니다.

{"verdict": "true", "statement_date": "6/11/2008", "statement_source": "speech"} 
{"verdict": "false", "statement_date": "6/7/2022", "statement_source": "television"}
{"verdict": "mostly-true", "statement_date": "5/18/2016", "statement_source": "news"}

Distributed Map 상태는 Amazon S3 버킷에 데이터 세트로 저장된 텍스트로 구분된 파일을 수락할 수 있습니다. 텍스트로 구분된 파일을 데이터 세트로 사용하는 경우 열 헤더를 지정해야 합니다. 헤더를 지정하는 방법에 대한 자세한 내용은 섹션을 참조하세요ItemReader 필드 콘텐츠.

Step Functions는 다음 규칙에 따라 텍스트로 구분된 파일을 구문 분석합니다.

Step Functions가 텍스트로 구분된 파일을 구문 분석하는 방법에 대한 자세한 내용은 섹션을 참조하세요Example of parsing an input CSV file.

워크플로 실행이 Map 상태에 도달하면 Step Functions는 GetObject API 작업을 호출하여 지정된 파일을 가져옵니다. 그런 다음 Map 상태는 파일의 각 행을 반복하고 하위 워크플로 실행을 시작하여 각 행의 항목을 처리합니다. 예를 들어 100개의 행이 포함된 텍스트로 구분된 파일을 입력으로 제공한다고 가정해 보겠습니다. 그러면 인터프리터에서 각 행을 Map 상태에 전달합니다. Map 상태는 헤더 행 다음부터 순차적으로 항목을 처리합니다.

다음 탭에서는 이 데이터 세트의 하위 워크플로 실행에 전달된 ItemReader 필드 구문과 입력의 예제를 보여줍니다.

Parquet 파일을 입력 소스로 사용할 수 있습니다. Amazon S3에 저장된 Apache Parquet 파일은 대규모로 효율적인 열 기반 데이터 처리를 제공합니다.

Parquet 파일을 사용하는 경우 다음 조건이 적용됩니다.

다음은를 Parquet으로 InputType 설정하기 위한 ASL 구성의 예입니다.

"ItemReader": {
   "Resource": "arn:aws:states:::s3:getObject",
   "ReaderConfig": {
      "InputType": "PARQUET"
   },
   "Arguments": {
      "Bucket": "amzn-s3-demo-bucket",
      "Key": "my-parquet-data-file-1.parquet"
   }
}

UNLOAD 쿼리 결과에서 생성된 Athena 매니페스트 파일을 사용하여 Map 상태에 대한 데이터 파일의 소스를 지정할 수 있습니다. ManifestType를 ATHENA_DATA로,를 CSV, JSONL또는 InputType로 설정합니다Parquet.

UNLOAD 쿼리를 실행할 때 Athena는 실제 데이터 객체 외에도 데이터 매니페스트 파일을 생성합니다. 매니페스트 파일은 데이터 파일의 구조화된 CSV 목록을 제공합니다. 매니페스트와 데이터 파일은 모두 Amazon S3의 Athena 쿼리 결과 위치에 저장됩니다.

UNLOAD (<YOUR_SELECT_QUERY>) TO 'S3_URI_FOR_STORING_DATA_OBJECT' WITH (format = 'JSON')

프로세스의 개념적 개요는 다음과 같습니다.

이 기능은 Athena에서 CSV, JSONL 및 Parquet 출력 형식을 처리할 수 있습니다. 단일 매니페스트 파일에서 참조되는 모든 객체는 동일한 InputType 형식이어야 합니다. UNLOAD 쿼리에서 내보낸 CSV 객체는 첫 번째 줄에 헤더를 포함하지 않습니다. 열 헤더를 제공해야 하는지 CSVHeaderLocation 여부를 참조하세요.

맵 컨텍스트에는 데이터 소스에 따라 처리를 사용자 지정할 수 $states.context.Map.Item.Source 있는 도 포함됩니다.

다음은 Athena 매니페스트를 사용하도록 ItemReader 구성된의 구성 예제입니다.

"ItemReader": {
   "Resource": "arn:aws:states:::s3:getObject",
   "ReaderConfig": {
      "ManifestType": "ATHENA_DATA",
      "InputType": "CSV | JSONL | PARQUET"
   },
   "Arguments": {
      "Bucket": "<S3_BUCKET_NAME>",
      "Key": "<S3_KEY_PREFIX><QUERY_ID>-manifest.csv"
   }
}

Distributed Map 상태는 Amazon S3 버킷에 저장된 Amazon S3 인벤토리 매니페스트 파일을 데이터 세트로 허용할 수 있습니다.

워크플로 실행이 Map 상태에 도달하면 Step Functions는 GetObject API 작업을 간접적으로 호출하여 지정된 Amazon S3 인벤토리 매니페스트 파일을 가져옵니다.

기본적으로 Map 상태는 인벤토리의 객체를 반복하여 Amazon S3 인벤토리 객체 메타데이터의 배열을 반환합니다.

ManifestType을 S3_INVENTORY로 지정하면 InputType을 지정할 수 없습니다.

다음은 CSV 형식의 인벤토리 파일에 대한 예제입니다. 이 파일에는 amzn-s3-demo-source-bucket이라는 Amazon S3 버킷에 저장되어 있는 csvDataset 및 imageDataset 객체가 포함되어 있습니다.

"amzn-s3-demo-source-bucket","csvDataset/","0","2022-11-16T00:27:19.000Z"
"amzn-s3-demo-source-bucket","csvDataset/titles.csv","3399671","2022-11-16T00:29:32.000Z"
"amzn-s3-demo-source-bucket","imageDataset/","0","2022-11-15T20:00:44.000Z"
"amzn-s3-demo-source-bucket","imageDataset/n02085620_10074.jpg","27034","2022-11-15T20:02:16.000Z"
...

Amazon S3 인벤토리 매니페스트 파일의 다음 예는 인벤토리 객체 메타데이터에 대한 CSV 헤더를 보여줍니다.

{
  "sourceBucket" : "amzn-s3-demo-source-bucket",
  "destinationBucket" : "arn:aws:s3:::amzn-s3-demo-inventory",
  "version" : "2016-11-30",
  "creationTimestamp" : "1668560400000",
  "fileFormat" : "CSV",
  "fileSchema" : "Bucket, Key, Size, LastModifiedDate",
  "files" : [ {
    "key" : "amzn-s3-demo-bucket/destination-prefix/data/20e55de8-9c21-45d4-99b9-46c732000228.csv.gz",
    "size" : 7300,
    "MD5checksum" : "a7ff4a1d4164c3cd55851055ec8f6b20"
  } ]
}

다음 탭에서는 이 데이터 세트의 하위 워크플로 실행에 전달된 ItemReader 필드 구문과 입력의 예제를 보여줍니다.