python - 无法使用 Airflow 连接到 Postgres 数据库

Question

我一直在尝试连接到 postgres 数据库并将一些示例数据插入到我的表中。我已经在网络控制台中创建了一个连接。

这是我尝试连接的脚本

insert_postgress = PostgresOperator(task_id='my_postgres_task',
                      sql="INSERT INTO xyz VALUES (3, 69, 'this is a test!');",
                      postgres_conn_id='my_postgres',
                      autocommit=True,
                      database="test",
                      dag=dag)
```

> And here is the error

```
>>> airflow test question my_postgres_task 11/12/2019
[2019-12-11 17:19:44,445] {__init__.py:51} INFO - Using executor SequentialExecutor
[2019-12-11 17:19:44,446] {dagbag.py:92} INFO - Filling up the DagBag from /mnt/c/dag
[2019-12-11 17:19:44,696] {taskinstance.py:630} INFO - Dependencies all met for <TaskInstance: question.my_postgres_task 2019-11-12T00:00:00+00:00 [None]>
[2019-12-11 17:19:44,704] {taskinstance.py:630} INFO - Dependencies all met for <TaskInstance: question.my_postgres_task 2019-11-12T00:00:00+00:00 [None]>
[2019-12-11 17:19:44,705] {taskinstance.py:841} INFO -
--------------------------------------------------------------------------------
[2019-12-11 17:19:44,705] {taskinstance.py:842} INFO - Starting attempt 1 of 2
[2019-12-11 17:19:44,705] {taskinstance.py:843} INFO -
--------------------------------------------------------------------------------
[2019-12-11 17:19:44,706] {taskinstance.py:862} INFO - Executing <Task(PostgresOperator): my_postgres_task> on 2019-11-12T00:00:00+00:00
[2019-12-11 17:19:44,715] {postgres_operator.py:62} INFO - Executing: INSERT INTO xyz VALUES (3, 69, 'this is a test!');
[2019-12-11 17:19:44,747] {connection.py:296} ERROR - No JSON object could be decoded
Traceback (most recent call last):
  File "/usr/local/lib/python2.7/dist-packages/airflow/models/connection.py", line 294, in extra_dejson
    obj = json.loads(self.extra)
  File "/usr/lib/python2.7/json/__init__.py", line 339, in loads
    return _default_decoder.decode(s)
  File "/usr/lib/python2.7/json/decoder.py", line 364, in decode
    obj, end = self.raw_decode(s, idx=_w(s, 0).end())
  File "/usr/lib/python2.7/json/decoder.py", line 382, in raw_decode
    raise ValueError("No JSON object could be decoded")
ValueError: No JSON object could be decoded
[2019-12-11 17:19:44,750] {connection.py:297} ERROR - Failed parsing the json for conn_id my_postgres
[2019-12-11 17:19:44,750] {base_hook.py:84} INFO - Using connection to: id: my_postgres. Host: localhost, Port: 5432, Schema: public, Login: postgres, Password: XXXXXXXX, extra: {}
```

Answer 1

检查连接对象上的 extra 字段。如果您正在使用它，请确保它是有效的 JSON。如果您不使用它，请确保它是一个正确的空字符串(可以通过 UI 保存空格，这会触发此错误)。