Skip to content

Commit cc19675

Browse files
authored
Add DuckDB support to the Amazon provider (#73301)
Run DuckDB queries against S3 without wiring up credentials in the Dag. Point DuckDBExecuteQueryOperator at a duckdb_aws connection and the Amazon provider brokers credentials from your AWS connection into the DuckDB session. The connection type is what selects the AWS hook, so the same operator serves local files and S3. This follows the precedent of redshift and athena, where the generic SQL operator picks up the right hook based on the redshift and athena connections. DuckDB is an optional extra of this provider.
1 parent 836a556 commit cc19675

14 files changed

Lines changed: 1078 additions & 20 deletions

File tree

‎dev/breeze/tests/test_selective_checks.py‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1130,7 +1130,8 @@ def assert_outputs_are_printed(expected_outputs: dict[str, str], stderr: str):
11301130
("providers/amazon/src/airflow/providers/amazon/provider.yaml",),
11311131
{
11321132
"selected-providers-list-as-string": "amazon apache.hive cncf.kubernetes "
1133-
"common.compat common.messaging common.sql databricks exasol ftp google http imap microsoft.azure "
1133+
"common.compat common.messaging common.sql databricks duckdb exasol ftp google http imap "
1134+
"microsoft.azure "
11341135
"mongo mysql openlineage postgres salesforce ssh teradata",
11351136
"all-python-versions": f"['{DEFAULT_PYTHON_MAJOR_MINOR_VERSION}']",
11361137
"all-python-versions-list-as-string": DEFAULT_PYTHON_MAJOR_MINOR_VERSION,
@@ -1156,7 +1157,7 @@ def assert_outputs_are_printed(expected_outputs: dict[str, str], stderr: str):
11561157
{
11571158
"description": "amazon...google",
11581159
"test_types": "Providers[amazon] Providers[apache.hive,cncf.kubernetes,"
1159-
"common.compat,common.messaging,common.sql,databricks,exasol,ftp,http,imap,"
1160+
"common.compat,common.messaging,common.sql,databricks,duckdb,exasol,ftp,http,imap,"
11601161
"microsoft.azure,mongo,mysql,openlineage,postgres,salesforce,ssh,teradata] "
11611162
"Providers[google]",
11621163
}
@@ -1202,7 +1203,8 @@ def assert_outputs_are_printed(expected_outputs: dict[str, str], stderr: str):
12021203
("providers/amazon/src/airflow/providers/amazon/file.py",),
12031204
{
12041205
"selected-providers-list-as-string": "amazon apache.hive cncf.kubernetes "
1205-
"common.compat common.messaging common.sql databricks exasol ftp google http imap microsoft.azure "
1206+
"common.compat common.messaging common.sql databricks duckdb exasol ftp google http imap "
1207+
"microsoft.azure "
12061208
"mongo mysql openlineage postgres salesforce ssh teradata",
12071209
"all-python-versions": f"['{DEFAULT_PYTHON_MAJOR_MINOR_VERSION}']",
12081210
"all-python-versions-list-as-string": DEFAULT_PYTHON_MAJOR_MINOR_VERSION,
@@ -1225,7 +1227,7 @@ def assert_outputs_are_printed(expected_outputs: dict[str, str], stderr: str):
12251227
{
12261228
"description": "amazon...google",
12271229
"test_types": "Providers[amazon] Providers[apache.hive,cncf.kubernetes,"
1228-
"common.compat,common.messaging,common.sql,databricks,exasol,ftp,http,imap,"
1230+
"common.compat,common.messaging,common.sql,databricks,duckdb,exasol,ftp,http,imap,"
12291231
"microsoft.azure,mongo,mysql,openlineage,postgres,salesforce,ssh,teradata] "
12301232
"Providers[google]",
12311233
}

‎providers/amazon/docs/index.rst‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -150,6 +150,7 @@ Dependent package
150150
`apache-airflow-providers-apache-hive <https://airflow.apache.org/docs/apache-airflow-providers-apache-hive>`_ ``apache.hive``
151151
`apache-airflow-providers-cncf-kubernetes <https://airflow.apache.org/docs/apache-airflow-providers-cncf-kubernetes>`_ ``cncf.kubernetes``
152152
`apache-airflow-providers-common-messaging <https://airflow.apache.org/docs/apache-airflow-providers-common-messaging>`_ ``common.messaging``
153+
`apache-airflow-providers-duckdb <https://airflow.apache.org/docs/apache-airflow-providers-duckdb>`_ ``duckdb``
153154
`apache-airflow-providers-exasol <https://airflow.apache.org/docs/apache-airflow-providers-exasol>`_ ``exasol``
154155
`apache-airflow-providers-ftp <https://airflow.apache.org/docs/apache-airflow-providers-ftp>`_ ``ftp``
155156
`apache-airflow-providers-google <https://airflow.apache.org/docs/apache-airflow-providers-google>`_ ``google``
@@ -194,6 +195,7 @@ Extra Dependencies
194195
``ssh`` ``apache-airflow-providers-ssh``
195196
``standard`` ``apache-airflow-providers-standard``
196197
``common.messaging`` ``apache-airflow-providers-common-messaging>=2.0.0``
198+
``duckdb`` ``apache-airflow-providers-duckdb>=0.1.0``
197199
``sqlalchemy`` ``sqlalchemy>=1.4.54,!=2.1.0``
198200
==================== ============================================================================================================================================================
199201

Lines changed: 180 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,180 @@
1+
.. Licensed to the Apache Software Foundation (ASF) under one
2+
or more contributor license agreements. See the NOTICE file
3+
distributed with this work for additional information
4+
regarding copyright ownership. The ASF licenses this file
5+
to you under the Apache License, Version 2.0 (the
6+
"License"); you may not use this file except in compliance
7+
with the License. You may obtain a copy of the License at
8+
9+
.. http://www.apache.org/licenses/LICENSE-2.0
10+
11+
.. Unless required by applicable law or agreed to in writing,
12+
software distributed under the License is distributed on an
13+
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
KIND, either express or implied. See the License for the
15+
specific language governing permissions and limitations
16+
under the License.
17+
18+
=============
19+
DuckDB on AWS
20+
=============
21+
22+
`DuckDB <https://duckdb.org/>`__ is an in-process analytical database. It can read and write data in
23+
Amazon S3 directly, which makes it an affordable way to run transforms that do not justify the scale
24+
of a distributed cluster.
25+
26+
DuckDB is not an AWS service, so the generic hook, operator and ``duckdb`` connection type live in the
27+
:doc:`DuckDB provider <apache-airflow-providers-duckdb:index>`. What this provider adds is the AWS
28+
auth. Using an Airflow AWS connection to create the DuckDB secret that grants access to S3.
29+
This means there is no credential wiring needed in the Dag.
30+
31+
You run :class:`~airflow.providers.duckdb.operators.duckdb.DuckDBExecuteQueryOperator` and point it
32+
at a ``duckdb_aws`` connection when the SQL should reach S3.
33+
34+
Prerequisite Tasks
35+
------------------
36+
37+
.. include:: ../_partials/prerequisite_tasks.rst
38+
39+
DuckDB support is an optional extra
40+
41+
.. code-block:: bash
42+
43+
pip install 'apache-airflow-providers-amazon[duckdb]'
44+
45+
.. note::
46+
47+
S3 access needs DuckDB's ``httpfs`` and ``aws`` extensions. This provider installs them if they
48+
are missing by default. If your workers should not download anything at runtime, set
49+
``autoinstall_extensions=False`` and make the extensions available yourself, either
50+
in an ``extension_directory`` or baked into your image.
51+
52+
.. _howto/connection:duckdb_aws:
53+
54+
Run a DuckDB query against Amazon S3
55+
====================================
56+
57+
Two connections are involved:
58+
59+
* A **DuckDB on AWS** (``duckdb_aws``) connection which configures DuckDB itself, i.e.: *which DuckDB database and with what settings*.
60+
Its id is what you pass to
61+
:class:`~airflow.providers.duckdb.operators.duckdb.DuckDBExecuteQueryOperator` as ``conn_id``, and its
62+
type is also what selects :class:`~airflow.providers.amazon.aws.hooks.duckdb.AwsDuckDBHook`.
63+
* An **AWS** (``aws``) connection says *which AWS identity*, i.e.: keys or role, region, endpoint, etc. The DuckDB
64+
connection points at it through ``aws_conn_id``, which defaults to ``aws_default``.
65+
66+
Credentials stay in the usual AWS connection. The ``duckdb_aws`` connection can be empty (though it must exist) and often
67+
will be with DuckDB's ability to use in-memory databases. Create it, name it whatever you like, and leave every field blank
68+
to get an in-memory database.
69+
70+
Given that connection, the hook loads the ``httpfs`` and ``aws`` extensions and creates the S3 secret,
71+
so the task supplies only SQL:
72+
73+
.. exampleinclude:: /../../amazon/tests/system/amazon/aws/example_duckdb.py
74+
:language: python
75+
:dedent: 4
76+
:start-after: [START howto_operator_aws_duckdb]
77+
:end-before: [END howto_operator_aws_duckdb]
78+
79+
Pass the id explicitly. With no ``conn_id`` the operator uses ``duckdb_default`` and the vendor-neutral
80+
hook, which has no S3 access.
81+
82+
The DuckDB connection carries the database and the engine settings; leave its ``Database path`` empty for
83+
an in-memory database, which suits a task that reads from S3 and writes back to it. Everything
84+
:class:`~airflow.providers.amazon.aws.hooks.duckdb.AwsDuckDBHook` accepts can be set in the connection
85+
``extra``, and ``hook_params`` on the operator overrides it per task. Set ``memory_limit`` and
86+
``threads`` explicitly: DuckDB otherwise sizes itself from the resources it detects on the host.
87+
88+
.. code-block:: python
89+
90+
hook_params = {"memory_limit": "2GB", "threads": 4}
91+
92+
.. note::
93+
94+
``aws_conn_id`` resolves on whether the key is present, not on its value, because ``None`` is a
95+
valid choice for it rather than the absence of one. Set it to ``null`` in the ``extra`` (or
96+
``None`` in ``hook_params``) to use no Airflow AWS connection at all and let the AWS SDK find
97+
credentials in the environment, such as environment variables or an instance role.
98+
99+
Using the hook directly
100+
=======================
101+
102+
For work that is not a single statement (e.g. chaining several queries against one database, or returning
103+
results to Python) you may use the hook directly. Opening one connection for several statements also
104+
avoids paying connection setup and extension loading more than once per task:
105+
106+
.. code-block:: python
107+
108+
@task
109+
def compare_regions():
110+
hook = AwsDuckDBHook(aws_conn_id="aws_default")
111+
with hook.get_conn() as conn:
112+
conn.execute("CREATE TABLE sales AS SELECT * FROM read_parquet('s3://my-bucket/sales.parquet')")
113+
return conn.execute("SELECT region, SUM(revenue) FROM sales GROUP BY region").fetchall()
114+
115+
How credentials reach DuckDB
116+
============================
117+
118+
DuckDB does not use ``boto3``. The hook therefore issues a ``CREATE SECRET`` statement that tells
119+
DuckDB how to authenticate, and the ``credential_strategy`` parameter selects how:
120+
121+
``credential_chain`` (default)
122+
Issues ``CREATE SECRET (TYPE s3, PROVIDER credential_chain)``, which resolves credentials through
123+
the AWS SDK inside DuckDB's ``aws`` extension. Nothing secret is written into SQL text and the SDK
124+
refreshes expiring credentials itself, so a long query does not fail partway through when a set of
125+
temporary credentials expires.
126+
127+
From DuckDB 1.4 onward the chain is resolved when the secret is created rather than when S3 is
128+
first read. If nothing can be resolved, the hook logs a warning and opens the connection anyway,
129+
so a query that does not use S3 still runs; one that does will fail on its own. A ``config``
130+
secret that cannot be created raises instead, since explicit credentials failing is a
131+
misconfiguration.
132+
133+
``config``
134+
Resolves credentials through
135+
:class:`~airflow.providers.amazon.aws.hooks.base_aws.AwsBaseHook` and writes them into the secret
136+
explicitly. This also works with a container-assigned role, because boto3 resolves that on the
137+
Airflow side, but the credentials are frozen when the connection is opened rather than refreshed.
138+
Needed when the Airflow connection carries static keys that the AWS SDK running inside
139+
DuckDB cannot see. The credentials become part of the SQL statement, so prefer
140+
``credential_chain`` where it works.
141+
142+
``none``
143+
Creates no secret. Use for a DuckDB database that never touches S3.
144+
145+
.. code-block:: python
146+
147+
DuckDBExecuteQueryOperator(
148+
task_id="query",
149+
conn_id="duckdb_aws_default",
150+
sql="SELECT COUNT(*) FROM read_parquet('s3://my-bucket/data.parquet')",
151+
hook_params={"aws_conn_id": "aws_static_keys", "credential_strategy": "config"},
152+
)
153+
154+
Non-default S3 endpoints
155+
========================
156+
157+
Set ``s3_endpoint_url``, or ``endpoint_url`` in the AWS connection extra, to reach an S3-compatible
158+
service or an S3 interface VPC endpoint. The hook translates it into the secret's ``ENDPOINT``,
159+
``URL_STYLE`` and ``USE_SSL`` clauses, selecting path-style addressing because virtual-host-style
160+
bucket names generally do not resolve against such endpoints.
161+
162+
.. note::
163+
164+
The S3 secret is the only channel through which Airflow configuration reaches DuckDB. A DuckDB
165+
session with no secret ignores the Airflow connection entirely — including its region and endpoint
166+
falling back to whatever DuckDB's own settings and environment provide.
167+
168+
DuckDB's ``httpfs`` extension reads credentials from the standard
169+
``AWS_ACCESS_KEY_ID`` / ``AWS_SECRET_ACCESS_KEY`` / ``AWS_SESSION_TOKEN`` environment
170+
variables, so it can authenticate where those are set. What its HTTP client does not implement is
171+
the ECS container credential provider, so a container-assigned task role (often how a managed Airflow
172+
worker receives credentials) is invisible to it and S3 access fails with an opaque
173+
``HTTP 403``. Both ``credential_chain`` and ``config`` handle that case; the difference is whether
174+
DuckDB or Airflow does the resolving, and therefore whether the credentials refresh.
175+
176+
Reference
177+
=========
178+
179+
* `DuckDB S3 API support <https://duckdb.org/docs/stable/extensions/httpfs/s3api>`__
180+
* `DuckDB secrets manager <https://duckdb.org/docs/stable/configuration/secrets_manager>`__

‎providers/amazon/provider.yaml‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -143,6 +143,14 @@ integrations:
143143
- /docs/apache-airflow-providers-amazon/operators/athena/athena_sql.rst
144144
- /docs/apache-airflow-providers-amazon/operators/athena/athena_spark.rst
145145
tags: [aws]
146+
# DuckDB itself is not an AWS service: the generic hook, operator and ``duckdb`` connection type
147+
# live in the ``duckdb`` provider. What lives here is only the AWS-specific part, brokering AWS
148+
# credentials and endpoints into a DuckDB session configured by its own connection type
149+
- integration-name: DuckDB on AWS
150+
external-doc-url: https://duckdb.org/docs/stable/extensions/httpfs/s3api
151+
how-to-guide:
152+
- /docs/apache-airflow-providers-amazon/operators/duckdb.rst
153+
tags: [aws]
146154
- integration-name: Amazon Bedrock
147155
external-doc-url: https://aws.amazon.com/bedrock/
148156
logo: /docs/integration-logos/Amazon-Bedrock_light-bg@4x.png
@@ -675,6 +683,9 @@ hooks:
675683
python-modules:
676684
- airflow.providers.amazon.aws.hooks.athena
677685
- airflow.providers.amazon.aws.hooks.athena_sql
686+
- integration-name: DuckDB on AWS
687+
python-modules:
688+
- airflow.providers.amazon.aws.hooks.duckdb
678689
- integration-name: Amazon Bedrock
679690
python-modules:
680691
- airflow.providers.amazon.aws.hooks.bedrock
@@ -1069,6 +1080,26 @@ connection-types:
10691080
schema: https
10701081
host: hooks.chime.aws/incomingwebhook/
10711082
password: T00000000?token=XXXXXXXXXXXXXXXXXXXXXXXX
1083+
- hook-class-name: airflow.providers.amazon.aws.hooks.duckdb.AwsDuckDBHook
1084+
hook-name: "DuckDB on AWS"
1085+
connection-type: duckdb_aws
1086+
ui-field-behaviour:
1087+
hidden-fields:
1088+
- login
1089+
- password
1090+
- port
1091+
- schema
1092+
relabeling:
1093+
host: Database path
1094+
placeholders:
1095+
host: /tmp/analytics.duckdb (leave empty for an in-memory database)
1096+
extra: |
1097+
{
1098+
"credential_strategy": "credential_chain",
1099+
"region_name": "us-east-1",
1100+
"memory_limit": "2GB",
1101+
"threads": 4
1102+
}
10721103
- hook-class-name: airflow.providers.amazon.aws.hooks.emr.EmrHook
10731104
hook-name: "Amazon Elastic MapReduce"
10741105
connection-type: emr

‎providers/amazon/pyproject.toml‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -152,6 +152,11 @@ dependencies = [
152152
"common.messaging" = [
153153
"apache-airflow-providers-common-messaging>=2.0.0"
154154
]
155+
# DuckDB support is opt-in: the duckdb wheel is around 20MB and irrelevant to most users of this
156+
# provider, so it must never become a hard dependency here.
157+
"duckdb" = [
158+
"apache-airflow-providers-duckdb>=0.1.0"
159+
]
155160
"sqlalchemy" = [
156161
"sqlalchemy>=1.4.54,!=2.1.0",
157162
]
@@ -166,6 +171,7 @@ dev = [
166171
"apache-airflow-providers-common-compat",
167172
"apache-airflow-providers-common-messaging",
168173
"apache-airflow-providers-common-sql",
174+
"apache-airflow-providers-duckdb",
169175
"apache-airflow-providers-exasol",
170176
"apache-airflow-providers-ftp",
171177
"apache-airflow-providers-google",
@@ -191,7 +197,8 @@ dev = [
191197
"opensearch-py>=2.2.0",
192198
"responses>=0.25.0",
193199
"apache-airflow-providers-common-sql[pandas,polars]",
194-
"apache-airflow-providers-amazon[sqlalchemy]"
200+
"apache-airflow-providers-amazon[sqlalchemy]",
201+
"apache-airflow-providers-duckdb"
195202
]
196203

197204
# To build docs:

0 commit comments

Comments
 (0)