使用密鑰AWS Secrets Manager阿帕奇氣流雪花連接

下列範例呼叫AWS Secrets Manager為 Apache 氣流亞馬遜管理工作流阿帕奇氣流雪花連接獲取秘密密鑰。它假設您已完成中的步驟使用 AWS Secrets Manager 密碼設定 Apache 氣流連線。

版本

• 您可以使用此頁面上的程式碼範例阿帕奇氣流 v2 及以上在蟒蛇。

先決條件

若要使用此頁面上的範例程式碼，您需要下列項目：

• 秘密管理器後端作為 Apache 氣流配置選項，如圖所示使用 AWS Secrets Manager 密碼設定 Apache 氣流連線。

• 秘密管理器中的 Apache 氣流連接字符串，如圖所示使用 AWS Secrets Manager 密碼設定 Apache 氣流連線。

許可

• 密碼管理員權限，如所示使用 AWS Secrets Manager 密碼設定 Apache 氣流連線。

請求

若要使用此頁面上的範例程式碼，請將下列相依性新增至requirements.txt。如需進一步了解，請參閱 安裝 Python 賴項。

apache-airflow-providers-snowflake==1.3.0

程式碼範例

下列步驟說明如何建立呼叫秘密管理員以取得密碼的 DAG 程式碼。

1. 在命令提示字元中，瀏覽至儲存 DAG 程式碼的目錄。例如：

   cd dags

2. 複製下列程式碼範例的內容，並在本機儲存為snowflake_connection.py。

   """
   Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
    
   Permission is hereby granted, free of charge, to any person obtaining a copy of
   this software and associated documentation files (the "Software"), to deal in
   the Software without restriction, including without limitation the rights to
   use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
   the Software, and to permit persons to whom the Software is furnished to do so.
    
   THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
   IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
   FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
   COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
   IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
   CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
   """
   from airflow import DAG
   from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator
   from airflow.utils.dates import days_ago
   
   snowflake_query = [
       """use warehouse "MY_WAREHOUSE";""",
       """select * from "SNOWFLAKE_SAMPLE_DATA"."WEATHER"."WEATHER_14_TOTAL" limit 100;""",
   ]
   
   with DAG(dag_id='snowflake_test', schedule_interval=None, catchup=False, start_date=days_ago(1)) as dag:
       snowflake_select = SnowflakeOperator(
           task_id="snowflake_select",
           sql=snowflake_query,
           snowflake_conn_id="snowflake_conn",
       )
   

後續步驟？

• 了解如何將此範例中的 DAG 程式碼上傳至dags亞馬遜 S3 存儲桶中的文件夾新增或更新 DAG。