Airflow配置Postgres后端时遭遇public schema权限拒绝问题

问题详情 #

已按Airflow官方文档完成以下操作：

• 创建airflow_db数据库
• 创建airflow_user用户并授予该用户数据库全部权限
• 修改airflow.cfg中的sql_alchemy_conn连接字符串为postgresql+psycopg2://airflow_user:***@localhost:5434/airflow_db

但执行airflow db init时，始终抛出权限错误，完整错误信息如下：

DB: postgresql+psycopg2://airflow_user:***@localhost:5434/airflow_db
[2022-11-10 15:19:38,319] {migration.py:204} INFO - Context impl PostgresqlImpl.
[2022-11-10 15:19:38,319] {migration.py:207} INFO - Will assume transactional DDL.
Traceback (most recent call last):
  File "/Users/phil/airflowv/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1802, in _execute_context
    self.dialect.do_execute(
  File "/Users/phil/airflowv/lib/python3.10/site-packages/sqlalchemy/engine/default.py", line 719, in do_execute
    cursor.execute(statement, parameters)
psycopg2.errors.InsufficientPrivilege: permission denied for schema public
LINE 2: CREATE TABLE dag_code (
                     ^


The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/Users/phil/airflowv/bin/airflow", line 8, in <module>
    sys.exit(main())
  File "/Users/phil/airflowv/lib/python3.10/site-packages/airflow/__main__.py", line 39, in main
    args.func(args)
  File "/Users/phil/airflowv/lib/python3.10/site-packages/airflow/cli/cli_parser.py", line 52, in command
    return func(*args, **kwargs)
  File "/Users/phil/airflowv/lib/python3.10/site-packages/airflow/cli/commands/db_command.py", line 37, in initdb
    db.initdb()
  File "/Users/phil/airflowv/lib/python3.10/site-packages/airflow/utils/session.py", line 75, in wrapper
    return func(*args, session=session, **kwargs)
  File "/Users/phil/airflowv/lib/python3.10/site-packages/airflow/utils/db.py", line 683, in initdb
    _create_db_from_orm(session=session)
  File "/Users/phil/airflowv/lib/python3.10/site-packages/airflow/utils/db.py", line 666, in _create_db_from_orm
    Base.metadata.create_all(settings.engine)
  File "/Users/phil/airflowv/lib/python3.10/site-packages/sqlalchemy/sql/schema.py", line 4785, in create_all
    bind._run_ddl_visitor(
  File "/Users/phil/airflowv/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 3110, in _run_ddl_visitor
    conn._run_ddl_visitor(visitorcallable, element, **kwargs)
  File "/Users/phil/airflowv/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 2113, in _run_ddl_visitor
    visitorcallable(self.dialect, self, **kwargs).traverse_single(element)
  File "/Users/phil/airflowv/lib/python3.10/site-packages/sqlalchemy/sql/visitors.py", line 524, in traverse_single
    return meth(obj, **kw)
  File "/Users/phil/airflowv/lib/python3.10/site-packages/sqlalchemy/sql/ddl.py", line 846, in visit_metadata
    self.traverse_single(
  File "/Users/phil/airflowv/lib/python3.10/site-packages/sqlalchemy/sql/visitors.py", line 524, in traverse_single
    return meth(obj, **kw)
  File "/Users/phil/airflowv/lib/python3.10/site-packages/sqlalchemy/sql/ddl.py", line 890, in visit_table
    self.connection.execute(
  File "/Users/phil/airflowv/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1289, in execute
    return meth(self, multiparams, params, _EMPTY_EXECUTION_OPTS)
  File "/Users/phil/airflowv/lib/python3.10/site-packages/sqlalchemy/sql/ddl.py", line 77, in _execute_on_connection
    return connection._execute_ddl(
  File "/Users/phil/airflowv/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1381, in _execute_ddl
    ret = self._execute_context(
  File "/Users/phil/airflowv/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1845, in _execute_context
    self._handle_dbapi_exception(
  File "/Users/phil/airflowv/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 2026, in _handle_dbapi_exception
    util.raise_(
  File "/Users/phil/airflowv/lib/python3.10/site-packages/sqlalchemy/util/compat.py", line 207, in raise_
    raise exception
  File "/Users/phil/airflowv/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1802, in _execute_context
    self.dialect.do_execute(
  File "/Users/phil/airflowv/lib/python3.10/site-packages/sqlalchemy/engine/default.py", line 719, in do_execute
    cursor.execute(statement, parameters)
sqlalchemy.exc.ProgrammingError: (psycopg2.errors.InsufficientPrivilege) permission denied for schema public
LINE 2: CREATE TABLE dag_code (
                     ^

[SQL: 
CREATE TABLE dag_code (
    fileloc_hash BIGINT NOT NULL, 
    fileloc VARCHAR(2000) NOT NULL, 
    last_updated TIMESTAMP WITH TIME ZONE NOT NULL, 
    source_code TEXT NOT NULL, 
    CONSTRAINT dag_code_pkey PRIMARY KEY (fileloc_hash)
)

]

解决方案 #

1. 补充用户对public schema的权限 #

登录Postgres命令行控制台，执行以下SQL语句：

-- 切换到airflow_db数据库
\c airflow_db;
-- 授予airflow_user对public schema的全部权限
GRANT ALL PRIVILEGES ON SCHEMA public TO airflow_user;
-- 授予用户对schema下所有现有表和序列的权限
GRANT ALL PRIVILEGES ON ALL TABLES IN SCHEMA public TO airflow_user;
GRANT ALL PRIVILEGES ON ALL SEQUENCES IN SCHEMA public TO airflow_user;

2. 验证权限配置 #

执行完上述命令后，可通过以下SQL确认权限是否生效：

SELECT nspname, usename, privilege_type
FROM information_schema.schema_privileges
WHERE nspname = 'public' AND usename = 'airflow_user';

确保查询结果包含CREATE、USAGE等关键权限。

3. 重新初始化数据库 #

回到Airflow环境，重新运行初始化命令：

airflow db init

内容的提问来源于stack exchange，提问作者PandaPhi