AWS 資料轉換服務與協調工具

import boto3
import csv

s3 = boto3.client('s3')

def lambda_handler(event, context):
    # 從事件中取得檔案名稱
    file_name = event['Records'][0]['s3']['key']
    
    # 從 S3 下載檔案
    s3.download_file('source-bucket', file_name, '/tmp/' + file_name)
    
    # 開啟檔案並驗證其內容
    with open('/tmp/' + file_name, 'r') as file:
        reader = csv.reader(file)
        for row in reader:
            # 對每一列進行處理
            print(row)

    # 將檔案移至不同的儲存桶
    s3.copy_object(CopySource='source-bucket/' + file_name, Bucket='destination-bucket', Key=file_name)
    s3.delete_object(Bucket='source-bucket', Key=file_name)

    return {
        'statusCode': 200,
        'statusMessage': 'OK'
    }

from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

args = getResolvedOptions(sys.argv, ['JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# 從 S3 中讀取 CSV 資料
datasource = glueContext.create_dynamic_frame.from_catalog(database="my_database", table_name="my_table")

# 對資料進行轉換
transformed_data = ApplyMapping.apply(frame=datasource, mappings=[("column1", "string", "column1", "string")])

# 將轉換後的資料寫入 S3
glueContext.write_dynamic_frame.from_options(frame=transformed_data, connection_type="s3", connection_options={"path": "s3://my-bucket/output/"}, format="parquet")

job.commit()

import boto3

glue = boto3.client('glue')

# 定義工作流程
workflow_name = 'my_workflow'
response = glue.create_workflow(
    Name=workflow_name,
    Description='My workflow'
)

# 定義工作流程圖
workflow_graph = {
    'Nodes': [
        {
            'Type': 'Crawler',
            'Name': 'my_crawler',
            'CrawlerName': 'my_crawler_name'
        },
        {
            'Type': 'Job',
            'Name': 'my_etl_job',
            'JobName': 'my_etl_job_name'
        }
    ],
    'Edges': [
        {
            'Source': 'my_crawler',
            'Destination': 'my_etl_job'
        }
    ]
}

# 建立工作流程圖
response = glue.create_workflow_graph(
    Name=workflow_name,
    Graph=workflow_graph
)

#### 內容解密：
1. **建立 AWS Glue 使用者端**：使用 `boto3` 函式庫建立與 AWS Glue 的連線，以便操作 AWS Glue 資源。
2. **定義工作流程名稱和描述**：指定工作流程的名稱和描述，以便於識別和管理。
3. **建立工作流程**：使用 `create_workflow` 方法建立新的工作流程。
4. **定義工作流程圖**：描述工作流程中的節點（例如爬蟲和 ETL 作業）以及它們之間的邊（即執行順序）。
5. **建立工作流程圖**：使用 `create_workflow_graph` 方法將定義好的工作流程圖與工作流程關聯起來。

### AWS Step Functions：複雜工作流程

AWS Step Functions 是另一種用於協調大資料管線的服務。它允許您建立複雜的工作流程，可以與多個 AWS 服務整合。Step Functions 是無伺服器的，這意味著您不需要佈署或管理任何基礎設施，並且只需根據使用量付費。

#### 圖 3.6：AWS Step Functions 狀態機

下圖顯示了一個 Step Functions 狀態機的範例：

此狀態機執行以下步驟：
1. 當檔案上傳到特定的 Amazon S3 儲存桶時，CloudWatch 事件被觸發，並啟動狀態機，傳入包含新上傳檔案位置的 JSON 物件。
2. 第一步驟「處理傳入檔案」執行一個 Glue Python shell 作業，該作業接收上傳檔案的位置作為輸入，並處理傳入檔案（例如，將 CSV 轉換為 Parquet 格式）。Python 函式的輸出指示檔案處理是否成功，如果成功，還包括寫入 Parquet 檔案的 S3 路徑。此資訊包含在傳遞給下一步驟的 JSON 中。
3. 「作業是否成功？」步驟是選擇型別。它檢查傳遞給該步驟的 JSON 資料，如果 `jobStatus` 欄位設定為 `succeeded`，則分支到「執行 AWS Glue 爬蟲」。如果 `jobStatus` 欄位設定為 `failed`，則分支到「作業失敗」。

#### 程式碼範例：定義 AWS Step Functions 狀態機
```json
{
  "StartAt": "ProcessIncomingFile",
  "States": {
    "ProcessIncomingFile": {
      "Type": "Task",
      "Resource": "arn:aws:glue:REGION:ACCOUNT_ID:job/my_etl_job",
      "Next": "DidJobSucceed?"
    },
    "DidJobSucceed?": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.jobStatus",
          "StringEquals": "succeeded",
          "Next": "RunAWSGlueCrawler"
        },
        {
          "Variable": "$.jobStatus",
          "StringEquals": "failed",
          "Next": "JobFailed"
        }
      ]
    }
  }
}

#### 內容解密：
1. **定義狀態機起始步驟**：指定狀態機的起始步驟為「處理傳入檔案」。
2. **描述狀態機中的狀態**：定義狀態機中的每個狀態，包括任務型別、使用的資源以及下一步驟。
3. **使用選擇狀態進行條件分支**：根據前一步驟的輸出進行條件判斷，決定後續執行的步驟。
4. **整合 AWS Glue ETL 作業**：在狀態機中使用 AWS Glue ETL 作業進行資料處理。