6

Apache 1.9.0 を使用しています。スノーフレーク フック プラグインを作成しました。$AIRFLOW_HOME/plugins ディレクトリにフックを配置しました。

$AIRFLOW_HOME
  +--plugins
    +--snowflake_hook2.py

snowflake_hook2.py

# This is the base class for a plugin
from airflow.plugins_manager import AirflowPlugin

# This is necessary to expose the plugin in the Web interface
from flask import Blueprint
from flask_admin import BaseView, expose
from flask_admin.base import MenuLink

# This is the base hook for connecting to a database
from airflow.hooks.dbapi_hook import DbApiHook

# This is the snowflake provided Connector
import snowflake.connector

# This is the default python logging package
import logging

class SnowflakeHook2(DbApiHook):
    """
    Airflow Hook to communicate with Snowflake
    This is implemented as a Plugin
    """
    def __init__(self, connname_in='snowflake_default', db_in='default', wh_in='default', schema_in='default'):
        logging.info('# Connecting to {0}'.format(connname_in))
        self.conn_name_attr = 'snowflake_conn_id'
        self.connname = connname_in
        self.superconn = super().get_connection(self.connname) #gets the values from Airflow

        {SNIP - Connection stuff that works}
        self.cur = self.conn.cursor()

    def query(self,q,params=None):
        """From jmoney's db_wrapper allows return of a full list of rows(tuples)"""
        if params == None: #no Params, so no insertion
            self.cur.execute(q)
        else: #make the parameter substitution
            self.cur.execute(q,params)
        self.results = self.cur.fetchall()
        self.rowcount = self.cur.rowcount
        self.columnnames = [colspec[0] for colspec in self.cur.description]
        return self.results
    {SNIP - Other class functions}

class SnowflakePluginClass(AirflowPlugin):
    name = "SnowflakePluginModule"
    hooks = [SnowflakeHook2]
    operators = []

そこで、Airflows plugin_manager に print ステートメントをいくつか入れて、何が起こっているのかをより適切に処理できるようにしました。ウェブサーバーを再起動して airflow list_dags を実行すると、これらの行に「新しいモジュール名」が表示されました (エラーはありません)

SnowflakePluginModule [<class '__home__ubuntu__airflow__plugins_snowflake_hook2.SnowflakeHook2'>]
hook_module -  airflow.hooks.snowflakepluginmodule
INTEGRATING airflow.hooks.snowflakepluginmodule
snowflakepluginmodule <module 'airflow.hooks.snowflakepluginmodule'>

これはドキュメントの内容と一致しているため、DAG でこれを使用しても問題ありません。

from airflow import DAG
from airflow.hooks.snowflakepluginmodule import SnowflakeHook2 
from airflow.operators.python_operator import PythonOperator

しかし、ウェブはこのエラーをスローします

Broken DAG: [/home/ubuntu/airflow/dags/test_sf2.py] No module named 'airflow.hooks.snowflakepluginmodule'

だから問題は、私は何を間違っているのですか？または、バグを発見しましたか？

4

2 に答える 2

4

以下のようにインポートする必要があります。

from airflow import DAG
from airflow.hooks import SnowflakeHook2 
from airflow.operators.python_operator import PythonOperator

また

from airflow import DAG
from airflow.hooks.SnowflakePluginModule import SnowflakeHook2 
from airflow.operators.python_operator import PythonOperator
于 2018-06-23T00:09:40.243 に答える
0

気流がプラグインディレクトリ内のフォルダーを自動的に通過し、その下のすべてを実行するとは思いません。私がうまくセットアップした方法は、__init__.py各プラグイン クラスを含むプラグイン ディレクトリの下に を配置することです。Github の Astronomer プラグインを見てください。プラグインの設定方法の非常に良い例がいくつか提供されています。

特に、mysql プラグインの設定方法を見てください。

https://github.com/airflow-plugins/mysql_plugin

また、誰かが最新バージョンの気流の 1 つにスノーフレーク フックを組み込みました。

https://github.com/apache/incubator-airflow/blob/master/airflow/contrib/hooks/snowflake_hook.py

于 2018-04-20T12:49:49.127 に答える