Enable specify the hook on GenericTransfer

[mariotaddeucci]

Description

Actually the GenericTransfer gets the hook based on connection's config.

The cloud providers have multiple hooks with the same connection.

Specifying a parameter destination_hook that receive the Hook class, we can set which hook we want to use, turning easier create new SQL transfers just adding the method insert_rows on hooks.

This could replace the SqlToS3Operator, SQLToGoogleSheetsOperator and OracleToAzureDataLakeOperator for example.

Use case/motivation

Makes easier to create new SQL transfers for cloud providers by using GenericTransfer.

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[potiuk]

This is not possible, because Hooks when instantiated might be created with different init parameters.

Also I think the 'GenerlcTransfer` is not that useful and I think it adds unnecessary complexity and generalization where we already have everything needed to support the "generic transfer" by writing decorated tasks and using hooks.

I think this is much nicer and more "new Airflow-y" way of implementing transfer tasks using precisely the combination of Hooks and connections you need. It's much more explicit:

@task
def transfer_task():
   s3_hook = S3Hook('s3_conneciton', ....params here ...)
   gcs_hook = GCSHook('gcs_connection',.. params here ...)
   # use .upload()/download() or streaming capabilities of the hooks to perform the transfer (or whatever each hook provide).

I will convert it into discussion and would love to hear what others think, but I believe we should simply describe this as a valid pattern to do transfer rather than investing into GenericTransfer.

WDYT @BasPH @dstandish @ashb @kaxil @mik-laj @turbaszek @olchas (I think at some point of time you were involved in similar discussions).