airflow.providers.amazon.aws.transfers.s3_to_sql¶

类¶

S3ToSqlOperator

将数据从 S3 加载到 SQL 数据库中。

模块内容¶

class airflow.providers.amazon.aws.transfers.s3_to_sql.S3ToSqlOperator(*, s3_key, s3_bucket, table, parser, column_list=None, commit_every=1000, schema=None, sql_conn_id='sql_default', sql_hook_params=None, aws_conn_id='aws_default', **kwargs)[source]¶

Bases: airflow.models.BaseOperator

将数据从 S3 加载到 SQL 数据库中。

你需要提供一个解析器函数，该函数以文件名作为输入并返回一个行的可迭代对象。

另请参阅

有关如何使用此操作符的更多信息，请参阅指南： Amazon S3 到 SQL 传输操作符

参数：

schema (str | None) – SQL 数据库中特定 schema 的引用
table (str) – SQL 数据库中特定 table 的引用
s3_bucket (str) – 特定 S3 存储桶的引用
s3_key (str) – 特定 S3 key 的引用
sql_conn_id (str) – 特定 SQL 数据库的引用。必须是 DBApiHook 类型。
sql_hook_params (dict | None) – 要传递给底层 hook 的额外配置参数。应与所需的 hook 构造函数参数匹配。
aws_conn_id (str | None) – 特定 S3 / AWS 连接的引用
column_list (list[str] | None) – 用于 insert SQL 的列名列表。
commit_every (int) – 单次事务中插入的最大行数。设置为 0 表示在一次事务中插入所有行。
parser (Callable[[str], collections.abc.Iterable[collections.abc.Iterable]]) –
一个解析器函数，以文件路径作为输入并返回一个可迭代对象。例如，要使用按行生成行的 CSV 解析器，请传递以下函数
```
def parse_csv(filepath):
    import csv

    with open(filepath, newline="") as file:
        yield from csv.reader(file)
```