如何将 Airflow 配置为使用 Snowflake 作为后端

Question

我正在尝试将Apache Airflow配置为使用Snowflake作为后端数据库。 从理论上讲，它应该开箱即用，因为它使用SQLAlchemy作为ORM，它支持Snowflake。

我已经确认SqlAlchemy通过成功连接到我们的Snowflake帐户与Snowflake一起工作。

我已经使用 alldb 选项安装了 Airflow 使用 sudo pip install apache-airflow[alldbs]和气流.cfg文件中，我已经将sql_alchemy_conn设置设置为SqlAlchemy连接字符串，该字符串适用于SqlAlchemy的create_engine((调用的手动测试。

我在运行airflow initdb时收到以下错误消息

[2019-06-20 14:08:28,268] {__init__.py:51} INFO - Using executor LocalExecutor
DB: snowflake://MYUSER:***@myaccount.us-east-1/MYDATABASE/AIRFLOW?warehouse=LOAD_WH
[2019-06-20 14:08:28,756] {db.py:350} INFO - Creating tables
Traceback (most recent call last):
  File "/usr/local/bin/airflow", line 32, in <module>
    args.func(args)
  File "/usr/local/lib/python2.7/dist-packages/airflow/bin/cli.py", line 1096, in initdb
    db.initdb(settings.RBAC)
  File "/usr/local/lib/python2.7/dist-packages/airflow/utils/db.py", line 91, in initdb
    upgradedb()
  File "/usr/local/lib/python2.7/dist-packages/airflow/utils/db.py", line 358, in upgradedb
    command.upgrade(config, 'heads')
  File "/usr/local/lib/python2.7/dist-packages/alembic/command.py", line 254, in upgrade
    script.run_env()
  File "/usr/local/lib/python2.7/dist-packages/alembic/script/base.py", line 427, in run_env
    util.load_python_file(self.dir, 'env.py')
  File "/usr/local/lib/python2.7/dist-packages/alembic/util/pyfiles.py", line 81, in load_python_file
    module = load_module_py(module_id, path)
  File "/usr/local/lib/python2.7/dist-packages/alembic/util/compat.py", line 141, in load_module_py
    mod = imp.load_source(module_id, path, fp)
  File "/usr/local/lib/python2.7/dist-packages/airflow/migrations/env.py", line 92, in <module>
    run_migrations_online()
  File "/usr/local/lib/python2.7/dist-packages/airflow/migrations/env.py", line 82, in run_migrations_online
    compare_type=COMPARE_TYPE,
  File "<string>", line 8, in configure
  File "/usr/local/lib/python2.7/dist-packages/alembic/runtime/environment.py", line 812, in configure
    opts=opts
  File "/usr/local/lib/python2.7/dist-packages/alembic/runtime/migration.py", line 172, in configure
    return MigrationContext(dialect, connection, opts, environment_context)
  File "/usr/local/lib/python2.7/dist-packages/alembic/runtime/migration.py", line 111, in __init__
    self.impl = ddl.DefaultImpl.get_by_dialect(dialect)(
  File "/usr/local/lib/python2.7/dist-packages/alembic/ddl/impl.py", line 65, in get_by_dialect
    return _impls[dialect.name]
KeyError: 'snowflake'```

Answer 1

就我而言，我修复了将此代码添加到env.py以便alembic可以识别雪花驱动程序

的问题

from alembic.ddl.impl import DefaultImpl
class SnowflakeImpl(DefaultImpl):
    __dialect__ = 'snowflake'

参考： https://docs.snowflake.net/manuals/user-guide/sqlalchemy.html#alembic-support