Как предотвратить запись зашифрованной переменной воздушного потока? - PullRequest
Новая
       36

Как предотвратить запись зашифрованной переменной воздушного потока?

0 голосов
/ 21 ноября 2018

Я установил поток воздуха, в котором объявлена ​​зашифрованная переменная.Я использую BigQueryOperator .Я использую зашифрованную переменную в SQL, который подается в класс.Но воздушный поток регистрирует SQL после расшифровки переменной.Как я могу предотвратить это?

1 Ответ

0 голосов
/ 22 ноября 2018

К сожалению, нет встроенного способа добиться этого.

Возможный обходной путь - удаление строки self.log.info('Executing: %s', self.sql) в BigQueryOperator или создание нового оператора, наследующего BigQueryOperator, как показано ниже: 

class CustomBQOperator(BigQueryOperator):
    @apply_defaults
    def __init__(self, *args, **kwargs):
        super(CustomBQOperator).__init__(*args, **kwargs)

    def execute(self, context):
        if self.bq_cursor is None:

            hook = BigQueryHook(
                bigquery_conn_id=self.bigquery_conn_id,
                use_legacy_sql=self.use_legacy_sql,
                delegate_to=self.delegate_to)
            conn = hook.get_conn()
            self.bq_cursor = conn.cursor()
        self.bq_cursor.run_query(
            self.sql,
            destination_dataset_table=self.destination_dataset_table,
            write_disposition=self.write_disposition,
            allow_large_results=self.allow_large_results,
            flatten_results=self.flatten_results,
            udf_config=self.udf_config,
            maximum_billing_tier=self.maximum_billing_tier,
            maximum_bytes_billed=self.maximum_bytes_billed,
            create_disposition=self.create_disposition,
            query_params=self.query_params,
            labels=self.labels,
            schema_update_options=self.schema_update_options,
            priority=self.priority,
            time_partitioning=self.time_partitioning,
            api_resource_configs=self.api_resource_configs,
            cluster_fields=self.cluster_fields,
        )

И затем использовать это CustomBQOperator вместо BigQueryOperator

...