giambattista.bloisi
/
lot1-kickoff
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

initial stage

This commit is contained in:
Giambattista Bloisi 2024-03-26 14:20:45 +01:00
parent 75221b489d
commit fc5f884f4d
2 changed files with 12 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
airflow/dags/import_EOSC_catalog.py
View File

@ -40,14 +40,15 @@ default_args = {
default_args=default_args,
params={
"S3_CONN_ID": "s3_conn",
"OPENSEARCH_CONN_ID": "opensearch_default",
"EOSC_CATALOG_BUCKET": "eosc-portal-import"
},
tags=["lot1"],
)
def import_EOSC_catalog():
@task
def create_indexes():
conn = BaseHook.get_connection('opensearch_default')
def create_indexes(**kwargs):
conn = BaseHook.get_connection(kwargs["params"]["OPENSEARCH_CONN_ID"])
client = OpenSearch(
hosts=[{'host': conn.host, 'port': conn.port}],
http_auth=(conn.login, conn.password),
@ -98,7 +99,7 @@ def import_EOSC_catalog():
@task
def bulk_load(files: list[(str, str)], **kwargs):
conn = BaseHook.get_connection('opensearch_default')
conn = BaseHook.get_connection(kwargs["params"]["OPENSEARCH_CONN_ID"])
client = OpenSearch(
hosts=[{'host': conn.host, 'port': conn.port}],
http_auth=(conn.login, conn.password),
@ -138,8 +139,8 @@ def import_EOSC_catalog():
print(f"Bulk-inserted {len(succeeded)} items (streaming_bulk).")
@task
def close_indexes():
conn = BaseHook.get_connection('opensearch_default')
def close_indexes(**kwargs):
conn = BaseHook.get_connection(kwargs["params"]["OPENSEARCH_CONN_ID"])
client = OpenSearch(
hosts=[{'host': conn.host, 'port': conn.port}],
http_auth=(conn.login, conn.password),

11
airflow/dags/import_EOSC_graph.py
View File

@ -42,14 +42,15 @@ default_args = {
default_args=default_args,
params={
"S3_CONN_ID": "s3_conn",
"OPENSEARCH_CONN_ID": "opensearch_default",
"EOSC_CATALOG_BUCKET": "eosc-portal-import"
},
tags=["lot1"]
)
def import_EOSC_graph():
@task
def create_indexes():
conn = BaseHook.get_connection('opensearch_default')
def create_indexes(**kwargs):
conn = BaseHook.get_connection(kwargs["params"]["OPENSEARCH_CONN_ID"])
client = OpenSearch(
hosts=[{'host': conn.host, 'port': conn.port}],
http_auth=(conn.login, conn.password),
@ -123,7 +124,7 @@ def import_EOSC_graph():
)
})
def bulk_load(files: list[(str, str)], **kwargs):
conn = BaseHook.get_connection('opensearch_default')
conn = BaseHook.get_connection(kwargs["params"]["OPENSEARCH_CONN_ID"])
client = OpenSearch(
hosts=[{'host': conn.host, 'port': conn.port}],
http_auth=(conn.login, conn.password),
@ -182,8 +183,8 @@ def import_EOSC_graph():
print(f"Bulk-inserted {succeeded} items (streaming_bulk).")
@task
def close_indexes():
conn = BaseHook.get_connection('opensearch_default')
def close_indexes(**kwargs):
conn = BaseHook.get_connection(kwargs["params"]["OPENSEARCH_CONN_ID"])
client = OpenSearch(
hosts=[{'host': conn.host, 'port': conn.port}],
http_auth=(conn.login, conn.password),