Невозможно запустить географические функции в операторе воздушного потока Bigquery - PullRequest
Новая
       76

Невозможно запустить географические функции в операторе воздушного потока Bigquery

0 голосов
/ 13 апреля 2020

Я довольно плохо знаком с географическими данными и воздушным потоком, поэтому, пожалуйста, прости меня и спроси точность, если мой вопрос не ясен.

Я пытаюсь запустить DAG через воздушный поток (google composer) , чтобы прочитать данные из таблицы в указанном наборе данных c, преобразовать указанный столбец c в тип GEOGRAPHY и вывести результат в другую таблицу: 


PROJECT = os.getenv("GCP_PROJECT")

default_args = {
    "owner": "Airflow",
    "depends_on_past": False,
    "start_date": datetime(2020, 4, 1),
    "email": ["foobarred@gmail.com"],
    "email_on_failure": False,
    "email_on_retry": False,
    "retries": 5,
    "retry_delay": timedelta(minutes=1),
    # 'queue': 'bash_queue',
    # 'pool': 'backfill',
    # 'priority_weight': 10,
    # 'end_date': datetime(2016, 1, 1),
}



dag = DAG(DAG_ID, default_args=default_args, schedule_interval='@once ', catchup=False)



ingestion_query = f"""
            SELECT 
                id, 
                epsg, 
                SAFE.ST_GEOGFROMTEXT(geometry) as geometry, 
                CURRENT_TIMESTAMP() as ingested_at
            FROM dataset_raw.trees
            WHERE SAFE.ST_GEOGFROMTEXT(geometry) is not NULL
        """
with dag:
    etl_operator = BigQueryOperator(sql=ingestion_query,
                                    destination_dataset_table=f'{PROJECT}.dataset_clean.trees',
                                    write_disposition="WRITE_TRUNCATE",
                                    task_id=f"full_dump_trees")

Запрос на загрузку протестирован и работает из консоли bigquery.

Однако при запуске DAG происходит сбой со следующим сообщением об ошибке 

INFO - Job 150: Subtask full_dump_trees             SELECT 
INFO - Job 150: Subtask full_dump_trees                 id, 
INFO - Job 150: Subtask full_dump_trees                 epsg, 
INFO - Job 150: Subtask full_dump_trees                 SAFE.ST_GEOGFROMTEXT(geometry) as geometry, 
INFO - Job 150: Subtask full_dump_trees                 CURRENT_TIMESTAMP() as ingested_at
INFO - Job 150: Subtask full_dump_trees             FROM geodata_raw.trees
INFO - Job 150: Subtask full_dump_trees             WHERE SAFE.ST_GEOGFROMTEXT(geometry) is not NULL
[...]
ERROR - BigQuery job failed. Final error was: 
{'reason': 'invalidQuery', 'location': 'query', 
 'message': '5.37 - 5.46: Unrecognized function safe.st_geogfromtext\n[Try using standard SQL (https://cloud.google.com/bigquery/docs/reference/standard-sql/enabling-standard-sql)]'}. The job was: {'kind': 'bigquery#job', 'etag': 'jBrdHGpAuAGJ48Z5A/ALIA==', 'id': 'strange-terra-273917:EU.job_oDUB9c3kz0NE1JKJ5tCaeHmuGN3d', 'selfLink': 'https://bigquery.googleapis.com/bigquery/v2/projects/strange-terra-273917/jobs/job_oDUB9c3kz0NE1JKJ5tCaeHmuGN3d?location=EU', 'user_email': 'foo@bar.iam.gserviceaccount.com', 'configuration': {'query': {'query': '\n            SELECT \n                id, \n                epsg, \n                SAFE.ST_GEOGFROMTEXT(geometry) as geometry, \n                CURRENT_TIMESTAMP() as ingested_at\n            FROM geodata_raw.trees\n            WHERE SAFE.ST_GEOGFROMTEXT(geometry) is not NULL\n        ', 'destinationTable': {'projectId': 'strange-terra-273917', 'datasetId': 'geodata_clean', 'tableId': 'trees'}, 'createDisposition': 'CREATE_IF_NEEDED', 'writeDisposition': 'WRITE_TRUNCATE', 'priority': 'INTERACTIVE', 'allowLargeResults': False, 'useLegacySql': True}, 'jobType': 'QUERY'}, 'jobReference': {'projectId': 'strange-terra-273917', 'jobId': 'job_oDUB9c3kz0NE1JKJ5tCaeHmuGN3d', 'location': 'EU'}, 'statistics': {'creationTime': '1586798059340', 'startTime': '1586798059356', 'endTime': '1586798059356'}, 'status': {'errorResult': {'reason': 'invalidQuery', 'location': 'query', 'message': '5.37 - 5.46: Unrecognized function safe.st_geogfromtext\n[Try using standard SQL (https://cloud.google.com/bigquery/docs/reference/standard-sql/enabling-standard-sql)]'}, 'errors': [{'reason': 'invalidQuery', 'location': 'query', 'message': '5.37 - 5.46: Unrecognized function safe.st_geogfromtext\n[Try using standard SQL (https://cloud.google.com/bigquery/docs/reference/standard-sql/enabling-standard-sql)]'}], 'state': 'DONE'}}
Traceback (most recent call last)
  File "/usr/local/lib/airflow/airflow/models/taskinstance.py", line 930, in _run_raw_tas
    result = task_copy.execute(context=context
  File "/usr/local/lib/airflow/airflow/contrib/operators/bigquery_operator.py", line 246, in execut
    encryption_configuration=self.encryption_configuratio
  File "/usr/local/lib/airflow/airflow/contrib/hooks/bigquery_hook.py", line 913, in run_quer
    return self.run_with_configuration(configuration
  File "/usr/local/lib/airflow/airflow/contrib/hooks/bigquery_hook.py", line 1344, in run_with_configuratio
    format(job['status']['errorResult'], job)
Exception: BigQuery job failed. Final error was: ...

Ошибка указывает на ссылку , которая кажется устаревшим. Документация, к которой он относится, похоже, говорит о том, что географические функции являются частью стандартного SQL, поэтому я затрудняюсь понять, почему это не сработает.

Это известное ограничение операторов больших запросов Airflow?

РЕДАКТИРОВАТЬ: Согласно документации , функция ST_GEOGFROMTEXT является частью того, что Google называет стандартным SQL для bigquery.

1 Ответ

3 голосов
/ 13 апреля 2020

Вы должны явно указать, что вы используете стандартный sql в своем коде - либо добавить # standard SQL в начало вашего запроса - убедитесь, что он находится в отдельной и первой строке вашего sql script
Или
Вы можете установить это в BigQueryOperator 

use_legacy_sql (bool) – Whether to use legacy SQL (true) or standard SQL (false).
...