airflow.providers.amazon.aws.transfers.s3_to_dynamodb¶

類¶

`AttributeDefinition`	屬性定義型別。
`KeySchema`	鍵模式型別。
`S3ToDynamoDBOperator`	將資料從 S3 載入到 DynamoDB。

模組內容¶

class airflow.providers.amazon.aws.transfers.s3_to_dynamodb.AttributeDefinition[source]¶

基類：TypedDict

屬性定義型別。

AttributeName: str[source]¶

AttributeType: Literal['S', 'N', 'B'][source]¶

class airflow.providers.amazon.aws.transfers.s3_to_dynamodb.KeySchema[source]¶

基類：TypedDict

鍵模式型別。

AttributeName: str[source]¶

KeyType: Literal['HASH', 'RANGE'][source]¶

class airflow.providers.amazon.aws.transfers.s3_to_dynamodb.S3ToDynamoDBOperator(*, s3_bucket, s3_key, dynamodb_table_name, dynamodb_key_schema, dynamodb_attributes=None, dynamodb_tmp_table_prefix='tmp', delete_on_error=False, use_existing_table=False, input_format='DYNAMODB_JSON', billing_mode='PAY_PER_REQUEST', import_table_kwargs=None, import_table_creation_kwargs=None, wait_for_completion=True, check_interval=30, max_attempts=240, aws_conn_id='aws_default', **kwargs)[source]¶

基類：airflow.models.BaseOperator

將資料從 S3 載入到 DynamoDB。

儲存在 S3 中的資料可以上傳到新的或現有的 DynamoDB。支援的檔案格式包括 CSV、DynamoDB JSON 和 Amazon ION。

引數：

s3_bucket (str) – 要匯入的 S3 儲存桶
s3_key (str) – 從 S3 匯入單個或多個物件的鍵字首
dynamodb_table_name (str) – 要建立的表的名稱
dynamodb_key_schema (list[KeySchema]) – 主鍵和排序鍵。每個元素代表一個主鍵屬性。AttributeName 是屬性的名稱。KeyType 是屬性的角色。有效值包括 HASH 或 RANGE
dynamodb_attributes (list[AttributeDefinition] | None) – 表的屬性名稱。AttributeName 是屬性的名稱，AttributeType 是屬性的資料型別。AttributeType 的有效值包括 S - 屬性型別為 String N - 屬性型別為 Number B - 屬性型別為 Binary
dynamodb_tmp_table_prefix (str) – 臨時 DynamoDB 表的字首
delete_on_error (bool) – 如果設定此項，則在新 DynamoDB 表匯入出錯時將其刪除
use_existing_table (bool) – 是否匯入到現有的非新建 DynamoDB 表。如果設定為 true，資料將首先載入到一個臨時 DynamoDB 表中（使用 AWS ImportTable 服務），然後分塊檢索到記憶體中並載入到目標表中。如果設定為 false，將建立一個新的 DynamoDB 表，並由 AWS ImportTable 服務批次載入 S3 資料。
input_format (Literal['CSV', 'DYNAMODB_JSON', 'ION']) – 匯入資料的格式。InputFormat 的有效值包括 CSV、DYNAMODB_JSON 或 ION
billing_mode (Literal['PROVISIONED', 'PAY_PER_REQUEST']) – 表的計費模式。有效值包括 PROVISIONED 或 PAY_PER_REQUEST
on_demand_throughput – 最大讀寫單元數量的額外選項
import_table_kwargs (dict[str, Any] | None) – 要傳遞的任何其他可選匯入表引數，例如 ClientToken、InputCompressionType 或 InputFormatOptions。請參閱：https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb/client/import_table.html
import_table_creation_kwargs (dict[str, Any] | None) – 要傳遞的任何其他可選匯入表建立引數，例如 ProvisionedThroughput、SSESpecification 或 GlobalSecondaryIndexes。請參閱：https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb/client/import_table.html
wait_for_completion (bool) – 是否等待叢集停止
check_interval (int) – 狀態檢查之間等待的秒數
max_attempts (int) – 檢查作業完成的最大嘗試次數
aws_conn_id (str | None) – AWS 連線詳情的引用

template_fields: collections.abc.Sequence[str] = ('s3_bucket', 's3_key', 'dynamodb_table_name', 'dynamodb_key_schema', 'dynamodb_attributes',...[source]¶

ui_color = '#e2e8f0'[source]¶

s3_bucket[source]¶

s3_key[source]¶

dynamodb_table_name[source]¶

dynamodb_attributes = None[source]¶

dynamodb_tmp_table_prefix = 'tmp'[source]¶

delete_on_error = False[source]¶

use_existing_table = False[source]¶

dynamodb_key_schema[source]¶

input_format = 'DYNAMODB_JSON'[source]¶

billing_mode = 'PAY_PER_REQUEST'[source]¶

import_table_kwargs = None[source]¶

import_table_creation_kwargs = None[source]¶

wait_for_completion = True[source]¶

check_interval = 30[source]¶

max_attempts = 240[source]¶

aws_conn_id = 'aws_default'[source]¶

property tmp_table_name[source]¶: 臨時表名。

execute(context)[source]¶

從 Airflow 執行 S3 到 DynamoDB 作業。

引數：: context (airflow.utils.context.Context) – 任務例項的當前上下文
返回值：: Amazon 資源編號 (ARN)
返回型別：: str