Remove cursor methods calls from GCSToBigQueryOperator

[kazanzhy]

closes: #10288

I've tried to remove warnings during GCSToBigQueryOperator as described in issue #10288.
As a result, I copied almost all code from bq_hook.create_external_table and bq_hook.run_load which made the operator's execute method very sophisticated.

That's why I just switched from cursor method calls to BigQueryHook methods. In my opinion, these methods should not be deprecated.

We need someplace for complex dict creation, like configuration for bq_hook.insert_job() or table_recource for bq_hook.create_empty_table(). In my opinion, better to keep this code in BigQueryHook and don't move to Operator.

[github-actions]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.

[kazanzhy]

Warnings in BQ hook was added by this commit
8b54919

[potiuk]

WDYT @turbaszek :) ?

[potiuk]

Rebased it to make sure it is still ok.

[turbaszek]

We need someplace for complex dict creation, like configuration for bq_hook.insert_job() or table_recource for bq_hook.create_empty_table(). In my opinion, better to keep this code in BigQueryHook and don't move to Operator.

Those dicts should be created by users. If we create them then we tend to miss some input arguments. Also in my opinion one input which is well described by Google docs is better than 20 arguments with copy pasted descriptions.

[kazanzhy]

@turbaszek
I assume that warning should be when the user uses hook directly but shouldn't be displayed if we use this method from operators.

[turbaszek]

@turbaszek
I assume that warning should be when the user uses hook directly but shouldn't be displayed if we use this method from operators.

Correct, and I believe we have separated warning in operators. Users can use both hook and operators and in both cases should be informed.

[kazanzhy]

So I returned warnings back to the BQ hook and suppressed them from the operator.
And seems it still closes an issue.

Also, I think there are left no usages of bq_hook.get_conn().cursor().
Probably it might be deleted in future

[github-actions]

The PR is likely OK to be merged with just subset of tests for default Python and Database versions without running the full matrix of tests, because it does not modify the core of Airflow. If the committers decide that the full tests matrix is needed, they will add the label 'full tests needed'. Then you should rebase to the latest main or amend the last commit of the PR, and push it with --force-with-lease.

[eladkal]

Regarding the create_external_table we already handled it in a diffrent operator - see #24363

@cswpy @kazanzhy maybe we can create a helper function to do this as we have more than 1 operator to use this functionality?

I assume we need something similar for the run_load ?

[kazanzhy]

Very good work with the #24363.

As I wrote earlier in this discussion it should be the place with all this routine dict creation.
I tried to do it in my work, but it's awfully inconvenient. It's much easier to use Pycharm and autocomplete than explore Google docs to understand where would be placed exactly this argument.

My previous idea was to keep it in hook, but probably we could think about the handler.
Recently I updated my data pipelines and must say that the combination of ClusterGenerator and DataprocCreateClusterOperator is fantastic.I think it's a good balance between flexibility and code overload.
I'm working on another PR and it's quite big. Maybe @cswpy has time?

[cswpy]

I'd love to work on this. But require more context on this issue. From my understanding, both run_load and create_external_table suffer from too many arguments. So we want to have a helper function for dictionary creation like configurations and table_resource? I am not following the design of "the combination of ClusterGenerator and DataprocCreateClusterOperator", can you elaborate?

[kazanzhy]

here's the part of my DAG

from airflow.providers.google.cloud.operators.dataproc import ClusterGenerator,DataprocCreateClusterOperator

with DAG(**dag_args, default_args=default_task_args) as dag:
    dataproc_cluster = ClusterGenerator(
        project_id=project_id,
        storage_bucket='gs://...',
        num_masters=3,
        master_machine_type='ns-standard-4',
        master_disk_size=1000,
        num_workers=160,
        worker_machine_type='ns-standard-2',
        worker_disk_size=100,
        num_preemptible_workers=0,
        image_version="2.0.28-debian10",
        auto_delete_ttl=3600,
        subnetwork_uri='default',
        properties={
            'dataproc:dataproc.logging.stackdriver.job.driver.enable': "true",
            'dataproc:dataproc.logging.stackdriver.job.yarn.container.enable': "true",
        },
    )
    create_dataproc_cluster = DataprocCreateClusterOperator(
        task_id='create_dataproc_cluster',
        project_id=project_id,
        cluster_name='on_demand_cluster',
        region=region,
        labels=labels,
        gcp_conn_id=gcp_connection_id,
        cluster_config=dataproc_cluster.make()
    )

This pattern allows to create a dict for the DataprocCreateClusterOperator or use some constructor which has a lot of parameters and shows them in IDE