Compare commits
20 Commits
exp360cust
...
Default
| Author | SHA1 | Date | |
|---|---|---|---|
| b3cc1e6819 | |||
| 80f0fc6ded | |||
| 4dafdd077a | |||
| 4be9e2f84b | |||
| 795ab32493 | |||
| 0645352856 | |||
| e0488e875f | |||
| 69eda12943 | |||
| 4df51791a0 | |||
| a82e537b5a | |||
| 7349a46331 | |||
| 708ca0c08e | |||
| e778e151b4 | |||
| 8bf5c6cd73 | |||
| d9af6afd4d | |||
| 95c6813aa4 | |||
| 21e03f99b9 | |||
| 821eea2c8a | |||
| 1a3a741297 | |||
| b455455c41 |
@@ -0,0 +1 @@
|
||||
{"entityType": "dataset", "name": "failed PAyments", "description": "", "type": "VIRTUAL_DATASET", "sub_type": "Standard", "sql": "SELECT * FROM \"exp360-customer\".failedpayments", "sqlContext": [], "references": {}, "path": ["Ocular", "Default", "failed PAyments"], "version": "v1", "created_date": "2026-07-23 09:01:56.244827", "updated_date": "2026-07-23 09:01:56.244827", "created_by": null, "auto_version": true, "snapshot_mode": true, "overwrite_mode": false, "snapshot_id": null, "schedule": null, "lineage": null, "status": "active", "schema_hash": "dee77168aec2"}
|
||||
@@ -0,0 +1,25 @@
|
||||
{
|
||||
"name": "payments dataset for test",
|
||||
"type": "VIRTUAL_DATASET",
|
||||
"sub_type": "Standard",
|
||||
"description": "",
|
||||
"latest_version": "v1",
|
||||
"versions": [
|
||||
{
|
||||
"version": "v1",
|
||||
"created_at": "2026-07-23T07:53:16.477381",
|
||||
"created_by": null,
|
||||
"schema_hash": "6a3ccf54a9b1",
|
||||
"status": "active",
|
||||
"is_current": true,
|
||||
"breaking_change": false,
|
||||
"change_notes": null,
|
||||
"dremio_view_name": "payments dataset for test"
|
||||
}
|
||||
],
|
||||
"schedule": null,
|
||||
"created_at": "2026-07-23T07:53:16.477381",
|
||||
"updated_at": "2026-07-23T07:53:16.477381",
|
||||
"created_by": null,
|
||||
"status": "active"
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
{"entityType": "dataset", "name": "payments dataset for test", "description": "", "type": "VIRTUAL_DATASET", "sub_type": "Standard", "sql": "SELECT * FROM \"exp360-customer\".payments", "sqlContext": [], "references": {}, "path": ["Ocular", "Default", "payments dataset for test"], "version": "v1", "created_date": "2026-07-23 07:53:16.477381", "updated_date": "2026-07-23 07:53:16.477381", "created_by": null, "auto_version": true, "snapshot_mode": true, "overwrite_mode": false, "snapshot_id": null, "schedule": null, "lineage": null, "status": "active", "schema_hash": "6a3ccf54a9b1"}
|
||||
@@ -0,0 +1 @@
|
||||
{"entityType": "dataset", "name": "playwright", "description": "", "type": "VIRTUAL_DATASET", "sub_type": "Standard", "sql": "SELECT * FROM \"exp360-customer\".actionsaudit", "sqlContext": [], "references": {}, "path": ["Ocular", "Default", "playwright"], "version": "v1", "created_date": "2026-07-23 07:35:35.566305", "updated_date": "2026-07-23 07:35:35.566305", "created_by": null, "auto_version": true, "snapshot_mode": true, "overwrite_mode": false, "snapshot_id": null, "schedule": null, "lineage": null, "status": "active", "schema_hash": "90b98cfa66e8"}
|
||||
@@ -0,0 +1 @@
|
||||
{"entityType": "dataset", "name": "screen visits dataset ", "description": "for test", "type": "VIRTUAL_DATASET", "sub_type": "Standard", "sql": "SELECT * FROM \"exp360-customer\".servicemetrics", "sqlContext": [], "references": {}, "path": ["Ocular", "Default", "screen visits dataset "], "version": "v1", "created_date": "2026-07-23 08:07:02.618969", "updated_date": "2026-07-23 08:07:02.618969", "created_by": null, "auto_version": true, "snapshot_mode": true, "overwrite_mode": false, "snapshot_id": null, "schedule": null, "lineage": null, "status": "active", "schema_hash": "d15f8d9f2633"}
|
||||
@@ -0,0 +1 @@
|
||||
{"entityType": "dataset", "name": "test DATA", "description": "", "type": "VIRTUAL_DATASET", "sub_type": "Standard", "sql": "SELECT * FROM \"exp360-customer\".failedpayments", "sqlContext": [], "references": {}, "path": ["Ocular", "Default", "test DATA"], "version": "v1", "created_date": "2026-07-23 07:39:04.531315", "updated_date": "2026-07-23 07:39:04.531315", "created_by": null, "auto_version": true, "snapshot_mode": true, "overwrite_mode": false, "snapshot_id": null, "schedule": null, "lineage": null, "status": "active", "schema_hash": "dee77168aec2"}
|
||||
@@ -0,0 +1 @@
|
||||
{"entityType": "dataset", "name": "test_dataset", "description": "testing", "type": "VIRTUAL_DATASET", "sub_type": "Standard", "sql": "SELECT * FROM \"exp360-customer\".actionsaudit limit 2;", "sqlContext": [], "references": {}, "path": ["Ocular", "Default", "test_dataset"], "version": "v1", "created_date": "2026-07-22 06:35:00.443268", "updated_date": "2026-07-22 06:35:00.443268", "created_by": null, "auto_version": true, "snapshot_mode": true, "overwrite_mode": false, "snapshot_id": null, "schedule": null, "lineage": null, "status": "active", "schema_hash": "f185dc8ab1f2"}
|
||||
25
__datasets/COD/Default/VIRTUAL_DATASET/test_v3/manifest.json
Normal file
25
__datasets/COD/Default/VIRTUAL_DATASET/test_v3/manifest.json
Normal file
@@ -0,0 +1,25 @@
|
||||
{
|
||||
"name": "test_v3",
|
||||
"type": "VIRTUAL_DATASET",
|
||||
"sub_type": "Standard",
|
||||
"description": "testing",
|
||||
"latest_version": "v1",
|
||||
"versions": [
|
||||
{
|
||||
"version": "v1",
|
||||
"created_at": "2026-07-22T06:19:55.134625",
|
||||
"created_by": null,
|
||||
"schema_hash": "f185dc8ab1f2",
|
||||
"status": "active",
|
||||
"is_current": true,
|
||||
"breaking_change": false,
|
||||
"change_notes": null,
|
||||
"dremio_view_name": "test_v3"
|
||||
}
|
||||
],
|
||||
"schedule": null,
|
||||
"created_at": "2026-07-22T06:19:55.134625",
|
||||
"updated_at": "2026-07-22T06:19:55.134625",
|
||||
"created_by": null,
|
||||
"status": "active"
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
{"entityType": "dataset", "name": "test_v3", "description": "testing", "type": "VIRTUAL_DATASET", "sub_type": "Standard", "sql": "SELECT * FROM \"exp360-customer\".actionsaudit limit 2;", "sqlContext": [], "references": {}, "path": ["Ocular", "Default", "test_v3"], "version": "v1", "created_date": "2026-07-22 06:19:55.134625", "updated_date": "2026-07-22 06:19:55.134625", "created_by": null, "auto_version": true, "snapshot_mode": true, "overwrite_mode": false, "snapshot_id": null, "schedule": null, "lineage": null, "status": "active", "schema_hash": "f185dc8ab1f2"}
|
||||
@@ -0,0 +1 @@
|
||||
{"entityType": "dataset", "name": "test_virtual_230726", "description": "", "type": "VIRTUAL_DATASET", "sub_type": "Standard", "sql": "SELECT * FROM \"exp360-customer\".actionsaudit limit 3;", "sqlContext": [], "references": {}, "path": ["Ocular", "Default", "test_virtual_230726"], "version": "v1", "created_date": "2026-07-23 14:50:22.925472", "updated_date": "2026-07-23 14:50:22.925472", "created_by": null, "auto_version": true, "snapshot_mode": true, "overwrite_mode": false, "snapshot_id": null, "schedule": null, "lineage": null, "status": "active", "schema_hash": "5a82ffa50310"}
|
||||
25
__datasets/COD/Default/VIRTUAL_DATASET/vds_12/manifest.json
Normal file
25
__datasets/COD/Default/VIRTUAL_DATASET/vds_12/manifest.json
Normal file
@@ -0,0 +1,25 @@
|
||||
{
|
||||
"name": "vds_12",
|
||||
"type": "VIRTUAL_DATASET",
|
||||
"sub_type": "Standard",
|
||||
"description": "test",
|
||||
"latest_version": "v1",
|
||||
"versions": [
|
||||
{
|
||||
"version": "v1",
|
||||
"created_at": "2026-07-23T16:43:36.855147",
|
||||
"created_by": null,
|
||||
"schema_hash": "f185dc8ab1f2",
|
||||
"status": "active",
|
||||
"is_current": true,
|
||||
"breaking_change": false,
|
||||
"change_notes": null,
|
||||
"dremio_view_name": "vds_12"
|
||||
}
|
||||
],
|
||||
"schedule": null,
|
||||
"created_at": "2026-07-23T16:43:36.855147",
|
||||
"updated_at": "2026-07-23T16:43:36.855147",
|
||||
"created_by": null,
|
||||
"status": "active"
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
{"entityType": "dataset", "name": "vds_12", "description": "test", "type": "VIRTUAL_DATASET", "sub_type": "Standard", "sql": "SELECT * FROM \"exp360-customer\".actionsaudit limit 2;", "sqlContext": [], "references": {}, "path": ["Ocular", "Default", "vds_12"], "version": "v1", "created_date": "2026-07-23 16:43:36.855147", "updated_date": "2026-07-23 16:43:36.855147", "created_by": null, "auto_version": true, "snapshot_mode": true, "overwrite_mode": false, "snapshot_id": null, "schedule": null, "lineage": null, "status": "active", "schema_hash": "f185dc8ab1f2"}
|
||||
25
__datasets/COD/Default/VIRTUAL_DATASET/vds_2/manifest.json
Normal file
25
__datasets/COD/Default/VIRTUAL_DATASET/vds_2/manifest.json
Normal file
@@ -0,0 +1,25 @@
|
||||
{
|
||||
"name": "vds_2",
|
||||
"type": "VIRTUAL_DATASET",
|
||||
"sub_type": "Standard",
|
||||
"description": "testing",
|
||||
"latest_version": "v1",
|
||||
"versions": [
|
||||
{
|
||||
"version": "v1",
|
||||
"created_at": "2026-07-23T16:47:44.492186",
|
||||
"created_by": null,
|
||||
"schema_hash": "1fe28ace3ab1",
|
||||
"status": "active",
|
||||
"is_current": true,
|
||||
"breaking_change": false,
|
||||
"change_notes": null,
|
||||
"dremio_view_name": "vds_2"
|
||||
}
|
||||
],
|
||||
"schedule": null,
|
||||
"created_at": "2026-07-23T16:47:44.492186",
|
||||
"updated_at": "2026-07-23T16:47:44.492186",
|
||||
"created_by": null,
|
||||
"status": "active"
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
{"entityType": "dataset", "name": "vds_2", "description": "testing", "type": "VIRTUAL_DATASET", "sub_type": "Standard", "sql": "SELECT * from \"exp360-customer\".actionsaudit limit 2;", "sqlContext": [], "references": {}, "path": ["Ocular", "Default", "vds_2"], "version": "v1", "created_date": "2026-07-23 16:47:44.492186", "updated_date": "2026-07-23 16:47:44.492186", "created_by": null, "auto_version": true, "snapshot_mode": true, "overwrite_mode": false, "snapshot_id": null, "schedule": null, "lineage": null, "status": "active", "schema_hash": "1fe28ace3ab1"}
|
||||
25
__datasets/COD/Default/VIRTUAL_DATASET/vds_3/manifest.json
Normal file
25
__datasets/COD/Default/VIRTUAL_DATASET/vds_3/manifest.json
Normal file
@@ -0,0 +1,25 @@
|
||||
{
|
||||
"name": "vds_3",
|
||||
"type": "VIRTUAL_DATASET",
|
||||
"sub_type": "Standard",
|
||||
"description": "testing",
|
||||
"latest_version": "v1",
|
||||
"versions": [
|
||||
{
|
||||
"version": "v1",
|
||||
"created_at": "2026-07-23T16:50:40.050859",
|
||||
"created_by": null,
|
||||
"schema_hash": "b06e066015a3",
|
||||
"status": "active",
|
||||
"is_current": true,
|
||||
"breaking_change": false,
|
||||
"change_notes": null,
|
||||
"dremio_view_name": "vds_3"
|
||||
}
|
||||
],
|
||||
"schedule": null,
|
||||
"created_at": "2026-07-23T16:50:40.050859",
|
||||
"updated_at": "2026-07-23T16:50:40.050859",
|
||||
"created_by": null,
|
||||
"status": "active"
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
{"entityType": "dataset", "name": "vds_3", "description": "testing", "type": "VIRTUAL_DATASET", "sub_type": "Standard", "sql": "select * from \"exp360-customer\".actionsaudit limit 2;", "sqlContext": [], "references": {}, "path": ["Ocular", "Default", "vds_3"], "version": "v1", "created_date": "2026-07-23 16:50:40.050859", "updated_date": "2026-07-23 16:50:40.050859", "created_by": null, "auto_version": true, "snapshot_mode": true, "overwrite_mode": false, "snapshot_id": null, "schedule": null, "lineage": null, "status": "active", "schema_hash": "b06e066015a3"}
|
||||
25
__datasets/COD/Default/VIRTUAL_DATASET/vds_4/manifest.json
Normal file
25
__datasets/COD/Default/VIRTUAL_DATASET/vds_4/manifest.json
Normal file
@@ -0,0 +1,25 @@
|
||||
{
|
||||
"name": "vds_4",
|
||||
"type": "VIRTUAL_DATASET",
|
||||
"sub_type": "Standard",
|
||||
"description": "testing",
|
||||
"latest_version": "v1",
|
||||
"versions": [
|
||||
{
|
||||
"version": "v1",
|
||||
"created_at": "2026-07-23T17:09:01.102143",
|
||||
"created_by": null,
|
||||
"schema_hash": "b06e066015a3",
|
||||
"status": "active",
|
||||
"is_current": true,
|
||||
"breaking_change": false,
|
||||
"change_notes": null,
|
||||
"dremio_view_name": "vds_4"
|
||||
}
|
||||
],
|
||||
"schedule": null,
|
||||
"created_at": "2026-07-23T17:09:01.102143",
|
||||
"updated_at": "2026-07-23T17:09:01.102143",
|
||||
"created_by": null,
|
||||
"status": "active"
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
{"entityType": "dataset", "name": "vds_4", "description": "testing", "type": "VIRTUAL_DATASET", "sub_type": "Standard", "sql": "select * from \"exp360-customer\".actionsaudit limit 2;", "sqlContext": [], "references": {}, "path": ["Ocular", "Default", "vds_4"], "version": "v1", "created_date": "2026-07-23 17:09:01.102143", "updated_date": "2026-07-23 17:09:01.102143", "created_by": null, "auto_version": true, "snapshot_mode": true, "overwrite_mode": false, "snapshot_id": null, "schedule": null, "lineage": null, "status": "active", "schema_hash": "b06e066015a3"}
|
||||
@@ -1,268 +0,0 @@
|
||||
|
||||
__generated_with = "0.13.15"
|
||||
|
||||
# %%
|
||||
|
||||
import sys
|
||||
sys.path.append('/opt/spark/work-dir/')
|
||||
from workflow_templates.spark.udf_manager import bootstrap_udfs
|
||||
from pyspark.sql.functions import udf
|
||||
from pyspark.sql.functions import lit
|
||||
from pyspark.sql.types import StringType, IntegerType
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
from pyspark import SparkConf, Row
|
||||
from pyspark.sql import SparkSession
|
||||
import os
|
||||
import pandas as pd
|
||||
import polars as pl
|
||||
import pyarrow as pa
|
||||
from pyspark.sql.functions import expr,to_json,col,struct
|
||||
from functools import reduce
|
||||
from handle_structs_or_arrays import preprocess_then_expand
|
||||
import requests
|
||||
from jinja2 import Template
|
||||
import json
|
||||
|
||||
|
||||
from secrets_manager import SecretsManager
|
||||
|
||||
from WorkflowManager import WorkflowDSL, WorkflowManager
|
||||
from KnowledgebaseManager import KnowledgebaseManager
|
||||
from gitea_client import GiteaClient, WorkspaceVersionedContent
|
||||
|
||||
from dremio.flight.endpoint import DremioFlightEndpoint
|
||||
from dremio.flight.query import DremioFlightEndpointQuery
|
||||
|
||||
|
||||
alias_str='abcdefghijklmnopqrstuvwxyz'
|
||||
workspace = os.getenv('WORKSPACE') or 'exp360cust'
|
||||
|
||||
job_id = os.getenv("EXECUTION_ID") or str(uuid.uuid4())
|
||||
|
||||
sm = SecretsManager(os.getenv('SECRET_MANAGER_URL'), os.getenv('SECRET_MANAGER_NAMESPACE'), os.getenv('SECRET_MANAGER_ENV'), os.getenv('SECRET_MANAGER_TOKEN'))
|
||||
secrets = sm.list_secrets(workspace)
|
||||
|
||||
gitea_client=GiteaClient(os.getenv('GITEA_HOST'), os.getenv('GITEA_TOKEN'), os.getenv('GITEA_OWNER') or 'gitea_admin', os.getenv('GITEA_REPO') or 'tenant1')
|
||||
workspaceVersionedContent=WorkspaceVersionedContent(gitea_client)
|
||||
conf = SparkConf()
|
||||
params = {
|
||||
"spark.hadoop.fs.s3a.access.key": secrets.get('S3_ACCESS_KEY'),
|
||||
"spark.hadoop.fs.s3a.secret.key": secrets.get('S3_SECRET_KEY'),
|
||||
"spark.hadoop.fs.s3a.aws.region": "us-west-1",
|
||||
"spark.sql.catalog.dremio.warehouse" : secrets.get('LAKEHOUSE_BUCKET'),
|
||||
"spark.sql.catalog.dremio" : "org.apache.iceberg.spark.SparkCatalog",
|
||||
"spark.sql.catalog.dremio.type" : "hadoop",
|
||||
"spark.hadoop.fs.s3a.impl": "org.apache.hadoop.fs.s3a.S3AFileSystem",
|
||||
"spark.hadoop.fs.s3.impl": "org.apache.hadoop.fs.s3a.S3AFileSystem",
|
||||
"spark.hadoop.fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem",
|
||||
"spark.sql.extensions": "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions",
|
||||
"spark.jars.packages": "com.amazonaws:aws-java-sdk-bundle:1.12.262,com.github.ben-manes.caffeine:caffeine:3.2.0,org.apache.iceberg:iceberg-aws-bundle:1.8.1,org.apache.iceberg:iceberg-common:1.8.1,org.apache.iceberg:iceberg-core:1.8.1,org.apache.iceberg:iceberg-spark:1.8.1,org.apache.hadoop:hadoop-aws:3.3.4,com.amazonaws:aws-java-sdk-bundle:1.11.901,org.apache.hadoop:hadoop-common:3.3.4,org.apache.hadoop:hadoop-cloud-storage:3.3.4,org.apache.hadoop:hadoop-client-runtime:3.3.4,org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.8.1,org.projectnessie.nessie-integrations:nessie-spark-extensions-3.5_2.12:0.103.2,org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.2,za.co.absa.cobrix:spark-cobol_2.12:2.8.0"
|
||||
}
|
||||
|
||||
|
||||
|
||||
conf.setAll(list(params.items()))
|
||||
|
||||
spark = SparkSession.builder.appName(workspace).config(conf=conf).getOrCreate()
|
||||
bootstrap_udfs(spark)
|
||||
|
||||
# %%
|
||||
|
||||
|
||||
|
||||
failed_payments_reader_df = spark.read.table('dremio.failedpayments')
|
||||
failed_payments_reader_df.createOrReplaceTempView('failed_payments_reader_df')
|
||||
|
||||
# %%
|
||||
|
||||
print(failed_payments_reader_df.columns)
|
||||
failed_payments_filter_df = spark.sql("select * from failed_payments_reader_df where retry_attempt_count < 3 AND gateway = \'CCS\' AND (retry_status = \'new\' OR retry_status = \'failed\')")
|
||||
failed_payments_filter_df.createOrReplaceTempView('failed_payments_filter_df')
|
||||
|
||||
# %%
|
||||
|
||||
|
||||
|
||||
|
||||
_payment_api_url = 'https://cod.uat.arconecloud.com/fw-notification/outbound-message-config/publish'
|
||||
|
||||
_payment_api_headers = dict()
|
||||
|
||||
for _payment_api_k,_payment_api_v in {'Content-Type': {'value': 'application/json', 'secret': None}, 'api-key': {'value': None, 'secret': 'API_KEY'}}.items() :
|
||||
|
||||
if(_payment_api_v.get('value') is not None and _payment_api_v.get('value') != ''):
|
||||
_payment_api_headers[_payment_api_k] = _payment_api_v.get('value')
|
||||
elif(_payment_api_v.get('secret') is not None and _payment_api_v.get('secret') != ''):
|
||||
_payment_api_headers[_payment_api_k] = secrets.get(_payment_api_v.get('secret'))
|
||||
|
||||
|
||||
|
||||
def _payment_api_call_api_udf(row):
|
||||
body_dict = row.asDict(recursive=True)
|
||||
template = Template('''{
|
||||
"msgData": {
|
||||
"canDistribute": true,
|
||||
"payment": [
|
||||
{
|
||||
"accountId3": "{{account_id}}",
|
||||
"currency2": "{{currency}}",
|
||||
"paymentAmount2": "{{amount}}"
|
||||
}
|
||||
],
|
||||
"paymentDate": "{{payment_date[:10]}}",
|
||||
"paymentTender": [
|
||||
{
|
||||
"currency3": "{{currency}}",
|
||||
"name": "NAME",
|
||||
"payorAccountId2": "{{account_id}}",
|
||||
"tenderAmount": "{{amount}}",
|
||||
"tenderType2": "CASH"
|
||||
}
|
||||
],
|
||||
"payorAccountId": "{{account_id}}",
|
||||
"shouldAllPaymentsFreeze": true,
|
||||
"tndrSrceCd": "CASH-01",
|
||||
"user": ""
|
||||
},
|
||||
"outMsgConfigCode": "EXP_CREATE_PAYMENT"
|
||||
}''')
|
||||
|
||||
body = json.loads(template.render(**body_dict))
|
||||
print("request : "+ json.dumps(body))
|
||||
|
||||
|
||||
response = requests.post(_payment_api_url, headers=_payment_api_headers, json=body, params={})
|
||||
|
||||
|
||||
data = response.json()
|
||||
print("response : " + json.dumps(data))
|
||||
|
||||
merged = {**body_dict, "response_body": data}
|
||||
return json.dumps(merged,default=str)
|
||||
|
||||
_payment_api_df_rdd=failed_payments_filter_df.rdd.map(lambda row: _payment_api_call_api_udf(row)).persist()
|
||||
payment_api_df = spark.read.json(_payment_api_df_rdd)
|
||||
if not "response_body" in payment_api_df.columns:
|
||||
raise Exception("Dataframe is empty")
|
||||
payment_api_df.createOrReplaceTempView('payment_api_df')
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
# %%
|
||||
|
||||
_failed_payments_update_mapper_select_clause=payment_api_df.columns if False else []
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("id AS id")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("retry_attempt_count + 1 AS retry_attempt_count")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("TRUE AS batch_processed")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("{job_id} AS batch_job_id")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("CASE WHEN response_body.success = \'true\' THEN \'success\' WHEN retry_attempt_count >= 3 THEN \'permanently_failed\' ELSE \'failed\' END AS retry_status")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("account_id AS account_id")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("failure_reason AS failure_reason")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("failure_timestamp AS failure_timestamp")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("CAST(amount AS DECIMAL(10,2)) AS amount")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("created_at AS created_at")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("payment_id AS payment_id")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("source_of_failure AS source_of_failure")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("currency AS currency")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("gateway AS gateway")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("payment_method AS payment_method")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("payment_date AS payment_date")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("date_format(current_timestamp(), \"yyyy-MM-dd\'T\'HH:mm:ss.SSS\") AS last_retry_timestamp")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("date_format(current_timestamp(), \"yyyy-MM-dd\'T\'HH:mm:ss.SSS\") AS updated_at")
|
||||
|
||||
|
||||
failed_payments_update_mapper_df=spark.sql(("SELECT " + ', '.join(_failed_payments_update_mapper_select_clause) + " FROM payment_api_df").replace("{job_id}",f"'{job_id}'"))
|
||||
failed_payments_update_mapper_df.createOrReplaceTempView("failed_payments_update_mapper_df")
|
||||
|
||||
# %%
|
||||
|
||||
print(payment_api_df.columns)
|
||||
success_payment_filter_df = spark.sql("select * from payment_api_df where response_body.success=\'true\'")
|
||||
success_payment_filter_df.createOrReplaceTempView('success_payment_filter_df')
|
||||
|
||||
# %%
|
||||
|
||||
_success_payments_mapper_select_clause=success_payment_filter_df.columns if False else []
|
||||
|
||||
_success_payments_mapper_select_clause.append("uuid() AS id")
|
||||
|
||||
_success_payments_mapper_select_clause.append("account_id AS account_id")
|
||||
|
||||
_success_payments_mapper_select_clause.append("currency AS currency")
|
||||
|
||||
_success_payments_mapper_select_clause.append("payment_id AS payment_id")
|
||||
|
||||
_success_payments_mapper_select_clause.append("gateway AS gateway")
|
||||
|
||||
_success_payments_mapper_select_clause.append("payment_method AS payment_method")
|
||||
|
||||
_success_payments_mapper_select_clause.append("payment_date AS payment_date")
|
||||
|
||||
_success_payments_mapper_select_clause.append("date_format(current_timestamp(), \"yyyy-MM-dd\'T\'HH:mm:ss.SSS\") AS created_at")
|
||||
|
||||
_success_payments_mapper_select_clause.append("date_format(current_timestamp(), \"yyyy-MM-dd\'T\'HH:mm:ss.SSS\") AS updated_at")
|
||||
|
||||
_success_payments_mapper_select_clause.append("CAST(amount AS DECIMAL(10,2)) AS amount")
|
||||
|
||||
|
||||
success_payments_mapper_df=spark.sql(("SELECT " + ', '.join(_success_payments_mapper_select_clause) + " FROM success_payment_filter_df").replace("{job_id}",f"'{job_id}'"))
|
||||
success_payments_mapper_df.createOrReplaceTempView("success_payments_mapper_df")
|
||||
|
||||
# %%
|
||||
|
||||
|
||||
|
||||
|
||||
_failed_payments_update_writer_fields_to_update = failed_payments_update_mapper_df.columns
|
||||
_failed_payments_update_writer_set_clause=[]
|
||||
_failed_payments_update_writer_unique_key_clause= []
|
||||
|
||||
for _key in ['payment_id']:
|
||||
_failed_payments_update_writer_unique_key_clause.append(f't.{_key} = s.{_key}')
|
||||
|
||||
for _field in _failed_payments_update_writer_fields_to_update:
|
||||
if(_field not in _failed_payments_update_writer_unique_key_clause):
|
||||
_failed_payments_update_writer_set_clause.append(f't.{_field} = s.{_field}')
|
||||
|
||||
_merge_query = '''
|
||||
MERGE INTO dremio.failedpayments t
|
||||
USING failed_payments_update_mapper_df s
|
||||
ON ''' + ' AND '.join(_failed_payments_update_writer_unique_key_clause) + ''' WHEN MATCHED THEN
|
||||
UPDATE SET ''' + ', '.join(_failed_payments_update_writer_set_clause) + ' WHEN NOT MATCHED THEN INSERT *'
|
||||
|
||||
spark.sql(_merge_query)
|
||||
|
||||
|
||||
|
||||
# %%
|
||||
|
||||
|
||||
|
||||
|
||||
success_payments_mapper_df.write.mode('append').saveAsTable('dremio.payments')
|
||||
|
||||
@@ -1,308 +0,0 @@
|
||||
import marimo
|
||||
|
||||
__generated_with = "0.13.15"
|
||||
app = marimo.App()
|
||||
|
||||
|
||||
@app.cell
|
||||
def init():
|
||||
|
||||
import sys
|
||||
sys.path.append('/opt/spark/work-dir/')
|
||||
from workflow_templates.spark.udf_manager import bootstrap_udfs
|
||||
from pyspark.sql.functions import udf
|
||||
from pyspark.sql.functions import lit
|
||||
from pyspark.sql.types import StringType, IntegerType
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
from pyspark import SparkConf, Row
|
||||
from pyspark.sql import SparkSession
|
||||
import os
|
||||
import pandas as pd
|
||||
import polars as pl
|
||||
import pyarrow as pa
|
||||
from pyspark.sql.functions import expr,to_json,col,struct
|
||||
from functools import reduce
|
||||
from handle_structs_or_arrays import preprocess_then_expand
|
||||
import requests
|
||||
from jinja2 import Template
|
||||
import json
|
||||
|
||||
|
||||
from secrets_manager import SecretsManager
|
||||
|
||||
from WorkflowManager import WorkflowDSL, WorkflowManager
|
||||
from KnowledgebaseManager import KnowledgebaseManager
|
||||
from gitea_client import GiteaClient, WorkspaceVersionedContent
|
||||
|
||||
from dremio.flight.endpoint import DremioFlightEndpoint
|
||||
from dremio.flight.query import DremioFlightEndpointQuery
|
||||
|
||||
|
||||
alias_str='abcdefghijklmnopqrstuvwxyz'
|
||||
workspace = os.getenv('WORKSPACE') or 'exp360cust'
|
||||
|
||||
job_id = os.getenv("EXECUTION_ID") or str(uuid.uuid4())
|
||||
|
||||
sm = SecretsManager(os.getenv('SECRET_MANAGER_URL'), os.getenv('SECRET_MANAGER_NAMESPACE'), os.getenv('SECRET_MANAGER_ENV'), os.getenv('SECRET_MANAGER_TOKEN'))
|
||||
secrets = sm.list_secrets(workspace)
|
||||
|
||||
gitea_client=GiteaClient(os.getenv('GITEA_HOST'), os.getenv('GITEA_TOKEN'), os.getenv('GITEA_OWNER') or 'gitea_admin', os.getenv('GITEA_REPO') or 'tenant1')
|
||||
workspaceVersionedContent=WorkspaceVersionedContent(gitea_client)
|
||||
conf = SparkConf()
|
||||
params = {
|
||||
"spark.hadoop.fs.s3a.access.key": secrets.get('S3_ACCESS_KEY'),
|
||||
"spark.hadoop.fs.s3a.secret.key": secrets.get('S3_SECRET_KEY'),
|
||||
"spark.hadoop.fs.s3a.aws.region": "us-west-1",
|
||||
"spark.sql.catalog.dremio.warehouse" : secrets.get('LAKEHOUSE_BUCKET'),
|
||||
"spark.sql.catalog.dremio" : "org.apache.iceberg.spark.SparkCatalog",
|
||||
"spark.sql.catalog.dremio.type" : "hadoop",
|
||||
"spark.hadoop.fs.s3a.impl": "org.apache.hadoop.fs.s3a.S3AFileSystem",
|
||||
"spark.hadoop.fs.s3.impl": "org.apache.hadoop.fs.s3a.S3AFileSystem",
|
||||
"spark.hadoop.fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem",
|
||||
"spark.sql.extensions": "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions",
|
||||
"spark.jars.packages": "com.amazonaws:aws-java-sdk-bundle:1.12.262,com.github.ben-manes.caffeine:caffeine:3.2.0,org.apache.iceberg:iceberg-aws-bundle:1.8.1,org.apache.iceberg:iceberg-common:1.8.1,org.apache.iceberg:iceberg-core:1.8.1,org.apache.iceberg:iceberg-spark:1.8.1,org.apache.hadoop:hadoop-aws:3.3.4,com.amazonaws:aws-java-sdk-bundle:1.11.901,org.apache.hadoop:hadoop-common:3.3.4,org.apache.hadoop:hadoop-cloud-storage:3.3.4,org.apache.hadoop:hadoop-client-runtime:3.3.4,org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.8.1,org.projectnessie.nessie-integrations:nessie-spark-extensions-3.5_2.12:0.103.2,org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.2,za.co.absa.cobrix:spark-cobol_2.12:2.8.0"
|
||||
}
|
||||
|
||||
|
||||
|
||||
conf.setAll(list(params.items()))
|
||||
|
||||
spark = SparkSession.builder.appName(workspace).config(conf=conf).getOrCreate()
|
||||
bootstrap_udfs(spark)
|
||||
return Template, job_id, json, requests, secrets, spark
|
||||
|
||||
|
||||
@app.cell
|
||||
def failed_payments_reader(spark):
|
||||
|
||||
|
||||
|
||||
failed_payments_reader_df = spark.read.table('dremio.failedpayments')
|
||||
failed_payments_reader_df.createOrReplaceTempView('failed_payments_reader_df')
|
||||
return (failed_payments_reader_df,)
|
||||
|
||||
|
||||
@app.cell
|
||||
def failed_payments_filter(failed_payments_reader_df, spark):
|
||||
|
||||
print(failed_payments_reader_df.columns)
|
||||
failed_payments_filter_df = spark.sql("select * from failed_payments_reader_df where retry_attempt_count < 3 AND gateway = \'CCS\' AND (retry_status = \'new\' OR retry_status = \'failed\')")
|
||||
failed_payments_filter_df.createOrReplaceTempView('failed_payments_filter_df')
|
||||
return (failed_payments_filter_df,)
|
||||
|
||||
|
||||
@app.cell
|
||||
def payment_api(
|
||||
Template,
|
||||
failed_payments_filter_df,
|
||||
json,
|
||||
requests,
|
||||
secrets,
|
||||
spark,
|
||||
):
|
||||
|
||||
|
||||
|
||||
|
||||
_payment_api_url = 'https://cod.uat.arconecloud.com/fw-notification/outbound-message-config/publish'
|
||||
|
||||
_payment_api_headers = dict()
|
||||
|
||||
for _payment_api_k,_payment_api_v in {'Content-Type': {'value': 'application/json', 'secret': None}, 'api-key': {'value': None, 'secret': 'API_KEY'}}.items() :
|
||||
|
||||
if(_payment_api_v.get('value') is not None and _payment_api_v.get('value') != ''):
|
||||
_payment_api_headers[_payment_api_k] = _payment_api_v.get('value')
|
||||
elif(_payment_api_v.get('secret') is not None and _payment_api_v.get('secret') != ''):
|
||||
_payment_api_headers[_payment_api_k] = secrets.get(_payment_api_v.get('secret'))
|
||||
|
||||
|
||||
|
||||
def _payment_api_call_api_udf(row):
|
||||
body_dict = row.asDict(recursive=True)
|
||||
template = Template('''{
|
||||
"msgData": {
|
||||
"canDistribute": true,
|
||||
"payment": [
|
||||
{
|
||||
"accountId3": "{{account_id}}",
|
||||
"currency2": "{{currency}}",
|
||||
"paymentAmount2": "{{amount}}"
|
||||
}
|
||||
],
|
||||
"paymentDate": "{{payment_date[:10]}}",
|
||||
"paymentTender": [
|
||||
{
|
||||
"currency3": "{{currency}}",
|
||||
"name": "NAME",
|
||||
"payorAccountId2": "{{account_id}}",
|
||||
"tenderAmount": "{{amount}}",
|
||||
"tenderType2": "CASH"
|
||||
}
|
||||
],
|
||||
"payorAccountId": "{{account_id}}",
|
||||
"shouldAllPaymentsFreeze": true,
|
||||
"tndrSrceCd": "CASH-01",
|
||||
"user": ""
|
||||
},
|
||||
"outMsgConfigCode": "EXP_CREATE_PAYMENT"
|
||||
}''')
|
||||
|
||||
body = json.loads(template.render(**body_dict))
|
||||
print("request : "+ json.dumps(body))
|
||||
|
||||
|
||||
response = requests.post(_payment_api_url, headers=_payment_api_headers, json=body, params={})
|
||||
|
||||
|
||||
data = response.json()
|
||||
print("response : " + json.dumps(data))
|
||||
|
||||
merged = {**body_dict, "response_body": data}
|
||||
return json.dumps(merged,default=str)
|
||||
|
||||
_payment_api_df_rdd=failed_payments_filter_df.rdd.map(lambda row: _payment_api_call_api_udf(row)).persist()
|
||||
payment_api_df = spark.read.json(_payment_api_df_rdd)
|
||||
if not "response_body" in payment_api_df.columns:
|
||||
raise Exception("Dataframe is empty")
|
||||
payment_api_df.createOrReplaceTempView('payment_api_df')
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
return (payment_api_df,)
|
||||
|
||||
|
||||
@app.cell
|
||||
def failed_payments_update_mapper(job_id, payment_api_df, spark):
|
||||
|
||||
_failed_payments_update_mapper_select_clause=payment_api_df.columns if False else []
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("id AS id")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("retry_attempt_count + 1 AS retry_attempt_count")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("TRUE AS batch_processed")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("{job_id} AS batch_job_id")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("CASE WHEN response_body.success = \'true\' THEN \'success\' WHEN retry_attempt_count >= 3 THEN \'permanently_failed\' ELSE \'failed\' END AS retry_status")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("account_id AS account_id")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("failure_reason AS failure_reason")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("failure_timestamp AS failure_timestamp")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("CAST(amount AS DECIMAL(10,2)) AS amount")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("created_at AS created_at")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("payment_id AS payment_id")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("source_of_failure AS source_of_failure")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("currency AS currency")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("gateway AS gateway")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("payment_method AS payment_method")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("payment_date AS payment_date")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("date_format(current_timestamp(), \"yyyy-MM-dd\'T\'HH:mm:ss.SSS\") AS last_retry_timestamp")
|
||||
|
||||
_failed_payments_update_mapper_select_clause.append("date_format(current_timestamp(), \"yyyy-MM-dd\'T\'HH:mm:ss.SSS\") AS updated_at")
|
||||
|
||||
|
||||
failed_payments_update_mapper_df=spark.sql(("SELECT " + ', '.join(_failed_payments_update_mapper_select_clause) + " FROM payment_api_df").replace("{job_id}",f"'{job_id}'"))
|
||||
failed_payments_update_mapper_df.createOrReplaceTempView("failed_payments_update_mapper_df")
|
||||
return (failed_payments_update_mapper_df,)
|
||||
|
||||
|
||||
@app.cell
|
||||
def success_payment_filter(payment_api_df, spark):
|
||||
|
||||
print(payment_api_df.columns)
|
||||
success_payment_filter_df = spark.sql("select * from payment_api_df where response_body.success=\'true\'")
|
||||
success_payment_filter_df.createOrReplaceTempView('success_payment_filter_df')
|
||||
return (success_payment_filter_df,)
|
||||
|
||||
|
||||
@app.cell
|
||||
def success_payments_mapper(job_id, spark, success_payment_filter_df):
|
||||
|
||||
_success_payments_mapper_select_clause=success_payment_filter_df.columns if False else []
|
||||
|
||||
_success_payments_mapper_select_clause.append("uuid() AS id")
|
||||
|
||||
_success_payments_mapper_select_clause.append("account_id AS account_id")
|
||||
|
||||
_success_payments_mapper_select_clause.append("currency AS currency")
|
||||
|
||||
_success_payments_mapper_select_clause.append("payment_id AS payment_id")
|
||||
|
||||
_success_payments_mapper_select_clause.append("gateway AS gateway")
|
||||
|
||||
_success_payments_mapper_select_clause.append("payment_method AS payment_method")
|
||||
|
||||
_success_payments_mapper_select_clause.append("payment_date AS payment_date")
|
||||
|
||||
_success_payments_mapper_select_clause.append("date_format(current_timestamp(), \"yyyy-MM-dd\'T\'HH:mm:ss.SSS\") AS created_at")
|
||||
|
||||
_success_payments_mapper_select_clause.append("date_format(current_timestamp(), \"yyyy-MM-dd\'T\'HH:mm:ss.SSS\") AS updated_at")
|
||||
|
||||
_success_payments_mapper_select_clause.append("CAST(amount AS DECIMAL(10,2)) AS amount")
|
||||
|
||||
|
||||
success_payments_mapper_df=spark.sql(("SELECT " + ', '.join(_success_payments_mapper_select_clause) + " FROM success_payment_filter_df").replace("{job_id}",f"'{job_id}'"))
|
||||
success_payments_mapper_df.createOrReplaceTempView("success_payments_mapper_df")
|
||||
return (success_payments_mapper_df,)
|
||||
|
||||
|
||||
@app.cell
|
||||
def failed_payments_update_writer(failed_payments_update_mapper_df, spark):
|
||||
|
||||
|
||||
|
||||
|
||||
_failed_payments_update_writer_fields_to_update = failed_payments_update_mapper_df.columns
|
||||
_failed_payments_update_writer_set_clause=[]
|
||||
_failed_payments_update_writer_unique_key_clause= []
|
||||
|
||||
for _key in ['payment_id']:
|
||||
_failed_payments_update_writer_unique_key_clause.append(f't.{_key} = s.{_key}')
|
||||
|
||||
for _field in _failed_payments_update_writer_fields_to_update:
|
||||
if(_field not in _failed_payments_update_writer_unique_key_clause):
|
||||
_failed_payments_update_writer_set_clause.append(f't.{_field} = s.{_field}')
|
||||
|
||||
_merge_query = '''
|
||||
MERGE INTO dremio.failedpayments t
|
||||
USING failed_payments_update_mapper_df s
|
||||
ON ''' + ' AND '.join(_failed_payments_update_writer_unique_key_clause) + ''' WHEN MATCHED THEN
|
||||
UPDATE SET ''' + ', '.join(_failed_payments_update_writer_set_clause) + ' WHEN NOT MATCHED THEN INSERT *'
|
||||
|
||||
spark.sql(_merge_query)
|
||||
|
||||
|
||||
return
|
||||
|
||||
|
||||
@app.cell
|
||||
def success_payments_writer(success_payments_mapper_df):
|
||||
|
||||
|
||||
|
||||
|
||||
success_payments_mapper_df.write.mode('append').saveAsTable('dremio.payments')
|
||||
|
||||
return
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
app.run()
|
||||
File diff suppressed because one or more lines are too long
@@ -1,842 +0,0 @@
|
||||
|
||||
__generated_with = "0.13.15"
|
||||
|
||||
# %%
|
||||
|
||||
import sys
|
||||
import time
|
||||
from pyspark.sql.utils import AnalysisException
|
||||
sys.path.append('/opt/spark/work-dir/')
|
||||
from workflow_templates.spark.udf_manager import bootstrap_udfs
|
||||
from util import get_logger, observe_metrics, collect_metrics, log_info, log_error, forgiving_serializer
|
||||
from pyspark.sql.functions import udf
|
||||
from pyspark.sql.functions import count, expr, lit
|
||||
from pyspark.sql.types import StringType, IntegerType
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
from pyspark import SparkConf, Row
|
||||
from pyspark.sql import SparkSession
|
||||
from pyspark.sql.observation import Observation
|
||||
from pyspark import StorageLevel
|
||||
import os
|
||||
import pandas as pd
|
||||
import polars as pl
|
||||
import pyarrow as pa
|
||||
from pyspark.sql.functions import approx_count_distinct, avg, collect_list, collect_set, corr, count, countDistinct, covar_pop, covar_samp, first, kurtosis, last, max, mean, min, skewness, stddev, stddev_pop, stddev_samp, sum, var_pop, var_samp, variance,expr,to_json,struct, date_format, col, lit, when, regexp_replace, ltrim
|
||||
from functools import reduce
|
||||
from handle_structs_or_arrays import preprocess_then_expand
|
||||
import requests
|
||||
from requests.adapters import HTTPAdapter
|
||||
from urllib3.util.retry import Retry
|
||||
from jinja2 import Template
|
||||
import json
|
||||
|
||||
|
||||
from secrets_manager import SecretsManager
|
||||
|
||||
from WorkflowManager import WorkflowDSL, WorkflowManager
|
||||
from KnowledgebaseManager import KnowledgebaseManager
|
||||
from gitea_client import GiteaClient, WorkspaceVersionedContent
|
||||
from FilesystemManager import FilesystemManager, SupportedFilesystemType
|
||||
from Materialization import Materialization
|
||||
|
||||
from dremio.flight.endpoint import DremioFlightEndpoint
|
||||
from dremio.flight.query import DremioFlightEndpointQuery
|
||||
|
||||
init_start_time=time.time()
|
||||
|
||||
LOGGER = get_logger()
|
||||
alias_str='abcdefghijklmnopqrstuvwxyz'
|
||||
workspace = os.getenv('WORKSPACE') or 'exp360cust'
|
||||
workflow = 'payment_metrics'
|
||||
execution_environment = os.getenv('EXECUTION_ENVIRONMENT') or 'CLUSTER'
|
||||
|
||||
job_id = os.getenv("EXECUTION_ID") or str(uuid.uuid4())
|
||||
retry_job_id = os.getenv("RETRY_EXECUTION_ID") or ''
|
||||
|
||||
log_info(LOGGER, f"Workspace: '{workspace}', Workflow: '{workflow}', Execution Environment: '{execution_environment}', Job Id: '{job_id}', Retry Job Id: '{retry_job_id}'")
|
||||
|
||||
sm = SecretsManager(os.getenv('SECRET_MANAGER_URL'), os.getenv('SECRET_MANAGER_NAMESPACE'), os.getenv('SECRET_MANAGER_ENV'), os.getenv('SECRET_MANAGER_TOKEN'))
|
||||
secrets = sm.list_secrets(workspace)
|
||||
|
||||
gitea_client=GiteaClient(os.getenv('GITEA_HOST'), os.getenv('GITEA_TOKEN'), os.getenv('GITEA_OWNER') or 'gitea_admin', os.getenv('GITEA_REPO') or 'tenant1')
|
||||
workspaceVersionedContent=WorkspaceVersionedContent(gitea_client)
|
||||
|
||||
filesystemManager = FilesystemManager.create(secrets.get('LAKEHOUSE_BUCKET'), storage_options={'key': secrets.get('S3_ACCESS_KEY'), 'secret': secrets.get('S3_SECRET_KEY'), 'region': secrets.get('S3_REGION')})
|
||||
if retry_job_id:
|
||||
logs = Materialization.get_execution_history_by_job_id(filesystemManager, secrets.get('LAKEHOUSE_BUCKET'), workspace, workflow, retry_job_id, selected_components=['finalize']).to_dicts()
|
||||
if len(logs) == 1 and logs[0].get('metrics').get('execute_status') == 'SUCCESS':
|
||||
log_info(LOGGER, f"Workspace: '{workspace}', Workflow: '{workflow}', Execution Environment: '{execution_environment}', Job Id: '{job_id}' - Retry Job Id: '{retry_job_id}' was already successful. Hence exiting to forward processing to next in chain.")
|
||||
sys.exit(0)
|
||||
|
||||
_conf = SparkConf()
|
||||
_params = {
|
||||
"spark.hadoop.fs.s3a.access.key": secrets.get('S3_ACCESS_KEY'),
|
||||
"spark.hadoop.fs.s3a.secret.key": secrets.get('S3_SECRET_KEY'),
|
||||
"spark.hadoop.fs.s3a.aws.region": secrets.get("S3_REGION") or "None",
|
||||
"spark.sql.catalog.dremio.warehouse" : secrets.get('LAKEHOUSE_BUCKET'),
|
||||
"spark.hadoop.fs.s3a.aws.credentials.provider": "com.amazonaws.auth.DefaultAWSCredentialsProviderChain",
|
||||
"spark.hadoop.fs.s3.aws.credentials.provider": "com.amazonaws.auth.DefaultAWSCredentialsProviderChain",
|
||||
"spark.sql.catalog.dremio" : "org.apache.iceberg.spark.SparkCatalog",
|
||||
"spark.sql.catalog.dremio.type" : "hadoop",
|
||||
"spark.hadoop.fs.s3a.impl": "org.apache.hadoop.fs.s3a.S3AFileSystem",
|
||||
"spark.hadoop.fs.s3.impl": "org.apache.hadoop.fs.s3a.S3AFileSystem",
|
||||
"spark.hadoop.fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem",
|
||||
"spark.sql.extensions": "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions",
|
||||
"spark.jars.packages": "com.amazonaws:aws-java-sdk-bundle:1.12.262,com.github.ben-manes.caffeine:caffeine:3.2.0,org.apache.iceberg:iceberg-aws-bundle:1.8.1,org.apache.iceberg:iceberg-common:1.8.1,org.apache.iceberg:iceberg-core:1.8.1,org.apache.iceberg:iceberg-spark:1.8.1,org.apache.hadoop:hadoop-aws:3.3.4,com.amazonaws:aws-java-sdk-bundle:1.11.901,org.apache.hadoop:hadoop-common:3.3.4,org.apache.hadoop:hadoop-cloud-storage:3.3.4,org.apache.hadoop:hadoop-client-runtime:3.3.4,org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.8.1,org.projectnessie.nessie-integrations:nessie-spark-extensions-3.5_2.12:0.103.2,org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.2,za.co.absa.cobrix:spark-cobol_2.12:2.8.0,ch.cern.sparkmeasure:spark-measure_2.12:0.26"
|
||||
}
|
||||
|
||||
|
||||
|
||||
_conf.setAll(list(_params.items()))
|
||||
|
||||
spark = SparkSession.builder.appName(workspace).config(conf=_conf).getOrCreate()
|
||||
bootstrap_udfs(spark)
|
||||
|
||||
materialization = Materialization(spark, secrets.get('LAKEHOUSE_BUCKET'), workspace, workflow, job_id, retry_job_id, execution_environment, LOGGER)
|
||||
|
||||
init_dependency_key="init"
|
||||
|
||||
|
||||
init_end_time=time.time()
|
||||
|
||||
# %%
|
||||
|
||||
|
||||
success_payments_reader_start_time=time.time()
|
||||
|
||||
success_payments_reader_fail_on_error=""
|
||||
try:
|
||||
success_payments_reader_df = spark.read.table('dremio.payments')
|
||||
success_payments_reader_df, success_payments_reader_observer = observe_metrics("success_payments_reader_df", success_payments_reader_df)
|
||||
|
||||
|
||||
|
||||
success_payments_reader_df.createOrReplaceTempView('success_payments_reader_df')
|
||||
|
||||
success_payments_reader_dependency_key="success_payments_reader"
|
||||
|
||||
success_payments_reader_execute_status="SUCCESS"
|
||||
except Exception as e:
|
||||
success_payments_reader_error = e
|
||||
log_error(LOGGER, f"Component success_payments_reader Failed", e)
|
||||
success_payments_reader_execute_status="ERROR"
|
||||
|
||||
raise e
|
||||
|
||||
finally:
|
||||
success_payments_reader_end_time=time.time()
|
||||
|
||||
# %%
|
||||
|
||||
|
||||
success_payments_mapper_start_time=time.time()
|
||||
|
||||
success_payments_mapper_fail_on_error=""
|
||||
try:
|
||||
_success_payments_mapper_select_clause=success_payments_reader_df.columns if False else []
|
||||
|
||||
_success_payments_mapper_select_clause.append('''DATE(payment_date) AS payment_date''')
|
||||
|
||||
_success_payments_mapper_select_clause.append('''amount AS amount''')
|
||||
|
||||
_success_payments_mapper_select_clause.append('''gateway AS gateway''')
|
||||
|
||||
_success_payments_mapper_select_clause.append('''payment_method AS payment_method''')
|
||||
|
||||
try:
|
||||
success_payments_mapper_df=spark.sql(("SELECT " + ', '.join(_success_payments_mapper_select_clause) + " FROM success_payments_reader_df").replace("{job_id}",f"'{job_id}'"))
|
||||
except Exception as e:
|
||||
success_payments_mapper_df = success_payments_reader_df.limit(0)
|
||||
log_info(LOGGER, f"error while mapping the data :{e} " )
|
||||
success_payments_mapper_df, success_payments_mapper_observer = observe_metrics("success_payments_mapper_df", success_payments_mapper_df)
|
||||
|
||||
|
||||
|
||||
success_payments_mapper_df.createOrReplaceTempView("success_payments_mapper_df")
|
||||
|
||||
success_payments_mapper_dependency_key="success_payments_mapper"
|
||||
|
||||
success_payments_mapper_execute_status="SUCCESS"
|
||||
except Exception as e:
|
||||
success_payments_mapper_error = e
|
||||
log_error(LOGGER, f"Component success_payments_mapper Failed", e)
|
||||
success_payments_mapper_execute_status="ERROR"
|
||||
|
||||
raise e
|
||||
|
||||
finally:
|
||||
success_payments_mapper_end_time=time.time()
|
||||
|
||||
# %%
|
||||
|
||||
|
||||
final_success_payments_start_time=time.time()
|
||||
|
||||
print(success_payments_mapper_df.columns)
|
||||
final_success_payments_fail_on_error=""
|
||||
try:
|
||||
try:
|
||||
final_success_payments_df = spark.sql("select * from success_payments_mapper_df where payment_date >= COALESCE((SELECT MAX(DATE(payment_date)) FROM dremio.successpaymentmetrics), (SELECT MIN(payment_date) FROM success_payments_mapper_df)) AND gateway = \'CCS\'")
|
||||
except AnalysisException as e:
|
||||
log_info(LOGGER, f"error while filtering data : {e}")
|
||||
final_success_payments_df = success_payments_mapper_df.limit(0)
|
||||
except Exception as e:
|
||||
log_info(LOGGER, f"Unexpected error: {e}")
|
||||
final_success_payments_df = success_payments_mapper_df.limit(0)
|
||||
final_success_payments_df, final_success_payments_observer = observe_metrics("final_success_payments_df", final_success_payments_df)
|
||||
|
||||
|
||||
|
||||
final_success_payments_df.createOrReplaceTempView('final_success_payments_df')
|
||||
|
||||
final_success_payments_dependency_key="final_success_payments"
|
||||
|
||||
final_success_payments_execute_status="SUCCESS"
|
||||
except Exception as e:
|
||||
final_success_payments_error = e
|
||||
log_error(LOGGER, f"Component final_success_payments Failed", e)
|
||||
final_success_payments_execute_status="ERROR"
|
||||
|
||||
raise e
|
||||
|
||||
finally:
|
||||
final_success_payments_end_time=time.time()
|
||||
|
||||
# %%
|
||||
|
||||
|
||||
high_valued_payments_filter_start_time=time.time()
|
||||
|
||||
print(final_success_payments_df.columns)
|
||||
high_valued_payments_filter_fail_on_error=""
|
||||
try:
|
||||
try:
|
||||
high_valued_payments_filter_df = spark.sql("select * from final_success_payments_df where amount >= 500")
|
||||
except AnalysisException as e:
|
||||
log_info(LOGGER, f"error while filtering data : {e}")
|
||||
high_valued_payments_filter_df = final_success_payments_df.limit(0)
|
||||
except Exception as e:
|
||||
log_info(LOGGER, f"Unexpected error: {e}")
|
||||
high_valued_payments_filter_df = final_success_payments_df.limit(0)
|
||||
high_valued_payments_filter_df, high_valued_payments_filter_observer = observe_metrics("high_valued_payments_filter_df", high_valued_payments_filter_df)
|
||||
|
||||
|
||||
|
||||
high_valued_payments_filter_df.createOrReplaceTempView('high_valued_payments_filter_df')
|
||||
|
||||
high_valued_payments_filter_dependency_key="high_valued_payments_filter"
|
||||
|
||||
high_valued_payments_filter_execute_status="SUCCESS"
|
||||
except Exception as e:
|
||||
high_valued_payments_filter_error = e
|
||||
log_error(LOGGER, f"Component high_valued_payments_filter Failed", e)
|
||||
high_valued_payments_filter_execute_status="ERROR"
|
||||
|
||||
raise e
|
||||
|
||||
finally:
|
||||
high_valued_payments_filter_end_time=time.time()
|
||||
|
||||
# %%
|
||||
|
||||
|
||||
total_payments_and_total_value_processed_start_time=time.time()
|
||||
|
||||
total_payments_and_total_value_processed_fail_on_error="True"
|
||||
try:
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
total_payments_and_total_value_processed_df = final_success_payments_df.groupBy(
|
||||
|
||||
"payment_date"
|
||||
).agg(
|
||||
|
||||
count('*').alias("total_payments"),
|
||||
|
||||
sum('amount').alias("total_value_processed")
|
||||
|
||||
)
|
||||
total_payments_and_total_value_processed_df, total_payments_and_total_value_processed_observer = observe_metrics("total_payments_and_total_value_processed_df", total_payments_and_total_value_processed_df)
|
||||
|
||||
|
||||
|
||||
total_payments_and_total_value_processed_df.createOrReplaceTempView('total_payments_and_total_value_processed_df')
|
||||
|
||||
total_payments_and_total_value_processed_dependency_key="total_payments_and_total_value_processed"
|
||||
|
||||
print(final_success_payments_dependency_key)
|
||||
|
||||
total_payments_and_total_value_processed_execute_status="SUCCESS"
|
||||
except Exception as e:
|
||||
total_payments_and_total_value_processed_error = e
|
||||
log_error(LOGGER, f"Component total_payments_and_total_value_processed Failed", e)
|
||||
total_payments_and_total_value_processed_execute_status="ERROR"
|
||||
|
||||
raise e
|
||||
|
||||
finally:
|
||||
total_payments_and_total_value_processed_end_time=time.time()
|
||||
|
||||
# %%
|
||||
|
||||
|
||||
aggregate__4_start_time=time.time()
|
||||
|
||||
aggregate__4_fail_on_error="True"
|
||||
try:
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
aggregate__4_df = final_success_payments_df.groupBy(
|
||||
|
||||
"payment_date",
|
||||
|
||||
"payment_method"
|
||||
).agg(
|
||||
|
||||
count('*').alias("method_count")
|
||||
|
||||
)
|
||||
aggregate__4_df, aggregate__4_observer = observe_metrics("aggregate__4_df", aggregate__4_df)
|
||||
|
||||
|
||||
|
||||
aggregate__4_df.createOrReplaceTempView('aggregate__4_df')
|
||||
|
||||
aggregate__4_dependency_key="aggregate__4"
|
||||
|
||||
print(final_success_payments_dependency_key)
|
||||
|
||||
aggregate__4_execute_status="SUCCESS"
|
||||
except Exception as e:
|
||||
aggregate__4_error = e
|
||||
log_error(LOGGER, f"Component aggregate__4 Failed", e)
|
||||
aggregate__4_execute_status="ERROR"
|
||||
|
||||
raise e
|
||||
|
||||
finally:
|
||||
aggregate__4_end_time=time.time()
|
||||
|
||||
# %%
|
||||
|
||||
|
||||
data_mapper__5_start_time=time.time()
|
||||
|
||||
data_mapper__5_fail_on_error=""
|
||||
try:
|
||||
_data_mapper__5_select_clause=aggregate__4_df.columns if False else []
|
||||
|
||||
_data_mapper__5_select_clause.append('''payment_date AS payment_date''')
|
||||
|
||||
_data_mapper__5_select_clause.append('''payment_method AS payment_method''')
|
||||
|
||||
_data_mapper__5_select_clause.append('''method_count AS method_count''')
|
||||
|
||||
_data_mapper__5_select_clause.append('''RANK() OVER (PARTITION BY payment_date ORDER BY method_count) AS rank_method''')
|
||||
|
||||
try:
|
||||
data_mapper__5_df=spark.sql(("SELECT " + ', '.join(_data_mapper__5_select_clause) + " FROM aggregate__4_df").replace("{job_id}",f"'{job_id}'"))
|
||||
except Exception as e:
|
||||
data_mapper__5_df = aggregate__4_df.limit(0)
|
||||
log_info(LOGGER, f"error while mapping the data :{e} " )
|
||||
data_mapper__5_df, data_mapper__5_observer = observe_metrics("data_mapper__5_df", data_mapper__5_df)
|
||||
|
||||
|
||||
|
||||
data_mapper__5_df.createOrReplaceTempView("data_mapper__5_df")
|
||||
|
||||
data_mapper__5_dependency_key="data_mapper__5"
|
||||
|
||||
data_mapper__5_execute_status="SUCCESS"
|
||||
except Exception as e:
|
||||
data_mapper__5_error = e
|
||||
log_error(LOGGER, f"Component data_mapper__5 Failed", e)
|
||||
data_mapper__5_execute_status="ERROR"
|
||||
|
||||
raise e
|
||||
|
||||
finally:
|
||||
data_mapper__5_end_time=time.time()
|
||||
|
||||
# %%
|
||||
|
||||
|
||||
filter__6_start_time=time.time()
|
||||
|
||||
print(data_mapper__5_df.columns)
|
||||
filter__6_fail_on_error=""
|
||||
try:
|
||||
try:
|
||||
filter__6_df = spark.sql("select * from data_mapper__5_df where rank_method = 1")
|
||||
except AnalysisException as e:
|
||||
log_info(LOGGER, f"error while filtering data : {e}")
|
||||
filter__6_df = data_mapper__5_df.limit(0)
|
||||
except Exception as e:
|
||||
log_info(LOGGER, f"Unexpected error: {e}")
|
||||
filter__6_df = data_mapper__5_df.limit(0)
|
||||
filter__6_df, filter__6_observer = observe_metrics("filter__6_df", filter__6_df)
|
||||
|
||||
|
||||
|
||||
filter__6_df.createOrReplaceTempView('filter__6_df')
|
||||
|
||||
filter__6_dependency_key="filter__6"
|
||||
|
||||
filter__6_execute_status="SUCCESS"
|
||||
except Exception as e:
|
||||
filter__6_error = e
|
||||
log_error(LOGGER, f"Component filter__6 Failed", e)
|
||||
filter__6_execute_status="ERROR"
|
||||
|
||||
raise e
|
||||
|
||||
finally:
|
||||
filter__6_end_time=time.time()
|
||||
|
||||
# %%
|
||||
|
||||
|
||||
most_used_payment_method___start_time=time.time()
|
||||
|
||||
most_used_payment_method___fail_on_error=""
|
||||
try:
|
||||
_most_used_payment_method___select_clause=filter__6_df.columns if False else []
|
||||
|
||||
_most_used_payment_method___select_clause.append('''payment_date AS payment_date''')
|
||||
|
||||
_most_used_payment_method___select_clause.append('''payment_method AS most_used_payment_method''')
|
||||
|
||||
try:
|
||||
most_used_payment_method___df=spark.sql(("SELECT " + ', '.join(_most_used_payment_method___select_clause) + " FROM filter__6_df").replace("{job_id}",f"'{job_id}'"))
|
||||
except Exception as e:
|
||||
most_used_payment_method___df = filter__6_df.limit(0)
|
||||
log_info(LOGGER, f"error while mapping the data :{e} " )
|
||||
most_used_payment_method___df, most_used_payment_method___observer = observe_metrics("most_used_payment_method___df", most_used_payment_method___df)
|
||||
|
||||
|
||||
|
||||
most_used_payment_method___df.createOrReplaceTempView("most_used_payment_method___df")
|
||||
|
||||
most_used_payment_method___dependency_key="most_used_payment_method__"
|
||||
|
||||
most_used_payment_method___execute_status="SUCCESS"
|
||||
except Exception as e:
|
||||
most_used_payment_method___error = e
|
||||
log_error(LOGGER, f"Component most_used_payment_method__ Failed", e)
|
||||
most_used_payment_method___execute_status="ERROR"
|
||||
|
||||
raise e
|
||||
|
||||
finally:
|
||||
most_used_payment_method___end_time=time.time()
|
||||
|
||||
# %%
|
||||
|
||||
|
||||
high_valued_payments___start_time=time.time()
|
||||
|
||||
high_valued_payments___fail_on_error="True"
|
||||
try:
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
high_valued_payments___df = high_valued_payments_filter_df.groupBy(
|
||||
|
||||
"payment_date"
|
||||
).agg(
|
||||
|
||||
count('*').alias("high_valued_payments")
|
||||
|
||||
)
|
||||
high_valued_payments___df, high_valued_payments___observer = observe_metrics("high_valued_payments___df", high_valued_payments___df)
|
||||
|
||||
|
||||
|
||||
high_valued_payments___df.createOrReplaceTempView('high_valued_payments___df')
|
||||
|
||||
high_valued_payments___dependency_key="high_valued_payments__"
|
||||
|
||||
print(high_valued_payments_filter_dependency_key)
|
||||
|
||||
high_valued_payments___execute_status="SUCCESS"
|
||||
except Exception as e:
|
||||
high_valued_payments___error = e
|
||||
log_error(LOGGER, f"Component high_valued_payments__ Failed", e)
|
||||
high_valued_payments___execute_status="ERROR"
|
||||
|
||||
raise e
|
||||
|
||||
finally:
|
||||
high_valued_payments___end_time=time.time()
|
||||
|
||||
# %%
|
||||
|
||||
|
||||
failed_payments_reader_start_time=time.time()
|
||||
|
||||
failed_payments_reader_fail_on_error=""
|
||||
try:
|
||||
failed_payments_reader_df = spark.read.table('dremio.failedpayments')
|
||||
failed_payments_reader_df, failed_payments_reader_observer = observe_metrics("failed_payments_reader_df", failed_payments_reader_df)
|
||||
|
||||
|
||||
|
||||
failed_payments_reader_df.createOrReplaceTempView('failed_payments_reader_df')
|
||||
|
||||
failed_payments_reader_dependency_key="failed_payments_reader"
|
||||
|
||||
failed_payments_reader_execute_status="SUCCESS"
|
||||
except Exception as e:
|
||||
failed_payments_reader_error = e
|
||||
log_error(LOGGER, f"Component failed_payments_reader Failed", e)
|
||||
failed_payments_reader_execute_status="ERROR"
|
||||
|
||||
raise e
|
||||
|
||||
finally:
|
||||
failed_payments_reader_end_time=time.time()
|
||||
|
||||
# %%
|
||||
|
||||
|
||||
failed_payments_mapper_start_time=time.time()
|
||||
|
||||
failed_payments_mapper_fail_on_error=""
|
||||
try:
|
||||
_failed_payments_mapper_select_clause=failed_payments_reader_df.columns if False else []
|
||||
|
||||
_failed_payments_mapper_select_clause.append('''DATE(payment_date) AS payment_date''')
|
||||
|
||||
_failed_payments_mapper_select_clause.append('''payment_method AS payment_method''')
|
||||
|
||||
_failed_payments_mapper_select_clause.append('''failure_reason AS failure_reason''')
|
||||
|
||||
_failed_payments_mapper_select_clause.append('''gateway AS gateway''')
|
||||
|
||||
try:
|
||||
failed_payments_mapper_df=spark.sql(("SELECT " + ', '.join(_failed_payments_mapper_select_clause) + " FROM failed_payments_reader_df").replace("{job_id}",f"'{job_id}'"))
|
||||
except Exception as e:
|
||||
failed_payments_mapper_df = failed_payments_reader_df.limit(0)
|
||||
log_info(LOGGER, f"error while mapping the data :{e} " )
|
||||
failed_payments_mapper_df, failed_payments_mapper_observer = observe_metrics("failed_payments_mapper_df", failed_payments_mapper_df)
|
||||
|
||||
|
||||
|
||||
failed_payments_mapper_df.createOrReplaceTempView("failed_payments_mapper_df")
|
||||
|
||||
failed_payments_mapper_dependency_key="failed_payments_mapper"
|
||||
|
||||
failed_payments_mapper_execute_status="SUCCESS"
|
||||
except Exception as e:
|
||||
failed_payments_mapper_error = e
|
||||
log_error(LOGGER, f"Component failed_payments_mapper Failed", e)
|
||||
failed_payments_mapper_execute_status="ERROR"
|
||||
|
||||
raise e
|
||||
|
||||
finally:
|
||||
failed_payments_mapper_end_time=time.time()
|
||||
|
||||
# %%
|
||||
|
||||
final_failed_payments_start_time=time.time()
|
||||
|
||||
print(failed_payments_mapper_df.columns)
|
||||
final_failed_payments_df = spark.sql("select * from failed_payments_mapper_df where payment_date >= COALESCE((SELECT MAX(DATE(payment_date)) FROM dremio.failedpaymentmetrics), (SELECT MIN(payment_date) FROM failed_payments_mapper_df))")
|
||||
final_failed_payments_df.createOrReplaceTempView('final_failed_payments_df')
|
||||
final_failed_payments_df.persist()
|
||||
|
||||
final_failed_payments_end_time=time.time()
|
||||
|
||||
final_failed_payments_dependency_key="final_failed_payments"
|
||||
|
||||
print(failed_payments_mapper_dependency_key)
|
||||
|
||||
|
||||
# %%
|
||||
|
||||
|
||||
filter__13_start_time=time.time()
|
||||
|
||||
print(final_failed_payments_df.columns)
|
||||
filter__13_fail_on_error=""
|
||||
try:
|
||||
try:
|
||||
filter__13_df = spark.sql("select * from final_failed_payments_df where gateway = \'CCS\'")
|
||||
except AnalysisException as e:
|
||||
log_info(LOGGER, f"error while filtering data : {e}")
|
||||
filter__13_df = final_failed_payments_df.limit(0)
|
||||
except Exception as e:
|
||||
log_info(LOGGER, f"Unexpected error: {e}")
|
||||
filter__13_df = final_failed_payments_df.limit(0)
|
||||
filter__13_df, filter__13_observer = observe_metrics("filter__13_df", filter__13_df)
|
||||
|
||||
|
||||
|
||||
filter__13_df.createOrReplaceTempView('filter__13_df')
|
||||
|
||||
filter__13_dependency_key="filter__13"
|
||||
|
||||
filter__13_execute_status="SUCCESS"
|
||||
except Exception as e:
|
||||
filter__13_error = e
|
||||
log_error(LOGGER, f"Component filter__13 Failed", e)
|
||||
filter__13_execute_status="ERROR"
|
||||
|
||||
raise e
|
||||
|
||||
finally:
|
||||
filter__13_end_time=time.time()
|
||||
|
||||
# %%
|
||||
|
||||
|
||||
total_failed_payments___start_time=time.time()
|
||||
|
||||
total_failed_payments___fail_on_error="True"
|
||||
try:
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
total_failed_payments___df = filter__13_df.groupBy(
|
||||
|
||||
"payment_date"
|
||||
).agg(
|
||||
|
||||
count('*').alias("total_failed_payments")
|
||||
|
||||
)
|
||||
total_failed_payments___df, total_failed_payments___observer = observe_metrics("total_failed_payments___df", total_failed_payments___df)
|
||||
|
||||
|
||||
|
||||
total_failed_payments___df.createOrReplaceTempView('total_failed_payments___df')
|
||||
|
||||
total_failed_payments___dependency_key="total_failed_payments__"
|
||||
|
||||
print(filter__13_dependency_key)
|
||||
|
||||
total_failed_payments___execute_status="SUCCESS"
|
||||
except Exception as e:
|
||||
total_failed_payments___error = e
|
||||
log_error(LOGGER, f"Component total_failed_payments__ Failed", e)
|
||||
total_failed_payments___execute_status="ERROR"
|
||||
|
||||
raise e
|
||||
|
||||
finally:
|
||||
total_failed_payments___end_time=time.time()
|
||||
|
||||
# %%
|
||||
|
||||
|
||||
failed_payment_metrics_start_time=time.time()
|
||||
|
||||
failed_payment_metrics_fail_on_error="True"
|
||||
try:
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
failed_payment_metrics_df = final_failed_payments_df.groupBy(
|
||||
|
||||
"payment_date",
|
||||
|
||||
"gateway",
|
||||
|
||||
"failure_reason"
|
||||
).agg(
|
||||
|
||||
count('*').alias("failure_count")
|
||||
|
||||
)
|
||||
failed_payment_metrics_df, failed_payment_metrics_observer = observe_metrics("failed_payment_metrics_df", failed_payment_metrics_df)
|
||||
|
||||
|
||||
|
||||
failed_payment_metrics_df.createOrReplaceTempView('failed_payment_metrics_df')
|
||||
|
||||
failed_payment_metrics_dependency_key="failed_payment_metrics"
|
||||
|
||||
print(final_failed_payments_dependency_key)
|
||||
|
||||
failed_payment_metrics_execute_status="SUCCESS"
|
||||
except Exception as e:
|
||||
failed_payment_metrics_error = e
|
||||
log_error(LOGGER, f"Component failed_payment_metrics Failed", e)
|
||||
failed_payment_metrics_execute_status="ERROR"
|
||||
|
||||
raise e
|
||||
|
||||
finally:
|
||||
failed_payment_metrics_end_time=time.time()
|
||||
|
||||
# %%
|
||||
|
||||
|
||||
data_writer__15_start_time=time.time()
|
||||
|
||||
data_writer__15_fail_on_error=""
|
||||
try:
|
||||
|
||||
_data_writer__15_fields_to_update = failed_payment_metrics_df.columns
|
||||
_data_writer__15_set_clause=[]
|
||||
_data_writer__15_unique_key_clause= []
|
||||
|
||||
for _key in ['payment_date', 'gateway', 'failure_reason']:
|
||||
_data_writer__15_unique_key_clause.append(f't.{_key} = s.{_key}')
|
||||
|
||||
for _field in _data_writer__15_fields_to_update:
|
||||
if(_field not in _data_writer__15_unique_key_clause):
|
||||
_data_writer__15_set_clause.append(f't.{_field} = s.{_field}')
|
||||
|
||||
_merge_query = '''
|
||||
MERGE INTO dremio.failedpaymentmetrics t
|
||||
USING failed_payment_metrics_df s
|
||||
ON ''' + ' AND '.join(_data_writer__15_unique_key_clause) + ''' WHEN MATCHED THEN
|
||||
UPDATE SET ''' + ', '.join(_data_writer__15_set_clause) + ' WHEN NOT MATCHED THEN INSERT *'
|
||||
|
||||
spark.sql(_merge_query)
|
||||
|
||||
|
||||
|
||||
data_writer__15_dependency_key="data_writer__15"
|
||||
|
||||
print(failed_payment_metrics_dependency_key)
|
||||
|
||||
data_writer__15_execute_status="SUCCESS"
|
||||
except Exception as e:
|
||||
data_writer__15_error = e
|
||||
log_error(LOGGER, f"Component data_writer__15 Failed", e)
|
||||
data_writer__15_execute_status="ERROR"
|
||||
|
||||
raise e
|
||||
|
||||
finally:
|
||||
data_writer__15_end_time=time.time()
|
||||
|
||||
# %%
|
||||
|
||||
success_payment_metrics_start_time=time.time()
|
||||
|
||||
print(total_payments_and_total_value_processed_df.columns)
|
||||
print(most_used_payment_method___df.columns)
|
||||
print(high_valued_payments___df.columns)
|
||||
print(total_failed_payments___df.columns)
|
||||
|
||||
success_payment_metrics_df = spark.sql("""
|
||||
SELECT
|
||||
COALESCE(a.payment_date, d.payment_date) AS payment_date,
|
||||
a.total_payments,
|
||||
a.total_value_processed,
|
||||
b.most_used_payment_method,
|
||||
c.high_valued_payments,
|
||||
d.total_failed_payments
|
||||
FROM total_failed_payments___df d
|
||||
FULL OUTER JOIN total_payments_and_total_value_processed_df a
|
||||
ON a.payment_date = d.payment_date
|
||||
LEFT JOIN most_used_payment_method___df b
|
||||
ON a.payment_date = b.payment_date
|
||||
LEFT JOIN high_valued_payments___df c
|
||||
ON a.payment_date = c.payment_date
|
||||
""")
|
||||
|
||||
success_payment_metrics_df.createOrReplaceTempView('success_payment_metrics_df')
|
||||
|
||||
success_payment_metrics_end_time=time.time()
|
||||
|
||||
success_payment_metrics_dependency_key="success_payment_metrics"
|
||||
|
||||
|
||||
# %%
|
||||
|
||||
|
||||
success_payment_metrics_writer_start_time=time.time()
|
||||
|
||||
success_payment_metrics_writer_fail_on_error=""
|
||||
try:
|
||||
|
||||
_success_payment_metrics_writer_fields_to_update = success_payment_metrics_df.columns
|
||||
_success_payment_metrics_writer_set_clause=[]
|
||||
_success_payment_metrics_writer_unique_key_clause= []
|
||||
|
||||
for _key in ['payment_date']:
|
||||
_success_payment_metrics_writer_unique_key_clause.append(f't.{_key} = s.{_key}')
|
||||
|
||||
for _field in _success_payment_metrics_writer_fields_to_update:
|
||||
if(_field not in _success_payment_metrics_writer_unique_key_clause):
|
||||
_success_payment_metrics_writer_set_clause.append(f't.{_field} = s.{_field}')
|
||||
|
||||
_merge_query = '''
|
||||
MERGE INTO dremio.successpaymentmetrics t
|
||||
USING success_payment_metrics_df s
|
||||
ON ''' + ' AND '.join(_success_payment_metrics_writer_unique_key_clause) + ''' WHEN MATCHED THEN
|
||||
UPDATE SET ''' + ', '.join(_success_payment_metrics_writer_set_clause) + ' WHEN NOT MATCHED THEN INSERT *'
|
||||
|
||||
spark.sql(_merge_query)
|
||||
|
||||
|
||||
|
||||
success_payment_metrics_writer_dependency_key="success_payment_metrics_writer"
|
||||
|
||||
success_payment_metrics_writer_execute_status="SUCCESS"
|
||||
except Exception as e:
|
||||
success_payment_metrics_writer_error = e
|
||||
log_error(LOGGER, f"Component success_payment_metrics_writer Failed", e)
|
||||
success_payment_metrics_writer_execute_status="ERROR"
|
||||
|
||||
raise e
|
||||
|
||||
finally:
|
||||
success_payment_metrics_writer_end_time=time.time()
|
||||
|
||||
# %%
|
||||
|
||||
finalize_start_time=time.time()
|
||||
|
||||
metrics = {
|
||||
'data': collect_metrics(locals()),
|
||||
}
|
||||
materialization.materialized_execution_history({'finalize': {'execute_status': 'SUCCESS', 'fail_on_error': 'False'}, **metrics['data']})
|
||||
log_info(LOGGER, f"Workflow Data metrics: {metrics['data']}")
|
||||
|
||||
finalize_end_time=time.time()
|
||||
|
||||
spark.stop()
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because one or more lines are too long
@@ -1,167 +0,0 @@
|
||||
|
||||
__generated_with = "0.13.15"
|
||||
|
||||
# %%
|
||||
|
||||
import sys
|
||||
sys.path.append('/opt/spark/work-dir/')
|
||||
from workflow_templates.spark.udf_manager import bootstrap_udfs
|
||||
from pyspark.sql.functions import udf
|
||||
from pyspark.sql.functions import lit
|
||||
from pyspark.sql.types import StringType, IntegerType
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
from pyspark import SparkConf, Row
|
||||
from pyspark.sql import SparkSession
|
||||
import os
|
||||
import pandas as pd
|
||||
import polars as pl
|
||||
import pyarrow as pa
|
||||
from pyspark.sql.functions import expr,to_json,col,struct
|
||||
from functools import reduce
|
||||
from handle_structs_or_arrays import preprocess_then_expand
|
||||
import requests
|
||||
from jinja2 import Template
|
||||
import json
|
||||
|
||||
|
||||
from secrets_manager import SecretsManager
|
||||
|
||||
from WorkflowManager import WorkflowDSL, WorkflowManager
|
||||
from KnowledgebaseManager import KnowledgebaseManager
|
||||
from gitea_client import GiteaClient, WorkspaceVersionedContent
|
||||
|
||||
from dremio.flight.endpoint import DremioFlightEndpoint
|
||||
from dremio.flight.query import DremioFlightEndpointQuery
|
||||
|
||||
|
||||
alias_str='abcdefghijklmnopqrstuvwxyz'
|
||||
workspace = os.getenv('WORKSPACE') or 'exp360cust'
|
||||
|
||||
job_id = os.getenv("EXECUTION_ID") or str(uuid.uuid4())
|
||||
|
||||
sm = SecretsManager(os.getenv('SECRET_MANAGER_URL'), os.getenv('SECRET_MANAGER_NAMESPACE'), os.getenv('SECRET_MANAGER_ENV'), os.getenv('SECRET_MANAGER_TOKEN'))
|
||||
secrets = sm.list_secrets(workspace)
|
||||
|
||||
gitea_client=GiteaClient(os.getenv('GITEA_HOST'), os.getenv('GITEA_TOKEN'), os.getenv('GITEA_OWNER') or 'gitea_admin', os.getenv('GITEA_REPO') or 'tenant1')
|
||||
workspaceVersionedContent=WorkspaceVersionedContent(gitea_client)
|
||||
conf = SparkConf()
|
||||
params = {
|
||||
"spark.hadoop.fs.s3a.access.key": secrets.get('S3_ACCESS_KEY'),
|
||||
"spark.hadoop.fs.s3a.secret.key": secrets.get('S3_SECRET_KEY'),
|
||||
"spark.hadoop.fs.s3a.aws.region": "us-west-1",
|
||||
"spark.sql.catalog.dremio.warehouse" : secrets.get('LAKEHOUSE_BUCKET'),
|
||||
"spark.sql.catalog.dremio" : "org.apache.iceberg.spark.SparkCatalog",
|
||||
"spark.sql.catalog.dremio.type" : "hadoop",
|
||||
"spark.hadoop.fs.s3a.impl": "org.apache.hadoop.fs.s3a.S3AFileSystem",
|
||||
"spark.hadoop.fs.s3.impl": "org.apache.hadoop.fs.s3a.S3AFileSystem",
|
||||
"spark.hadoop.fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem",
|
||||
"spark.sql.extensions": "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions",
|
||||
"spark.jars.packages": "com.amazonaws:aws-java-sdk-bundle:1.12.262,com.github.ben-manes.caffeine:caffeine:3.2.0,org.apache.iceberg:iceberg-aws-bundle:1.8.1,org.apache.iceberg:iceberg-common:1.8.1,org.apache.iceberg:iceberg-core:1.8.1,org.apache.iceberg:iceberg-spark:1.8.1,org.apache.hadoop:hadoop-aws:3.3.4,com.amazonaws:aws-java-sdk-bundle:1.11.901,org.apache.hadoop:hadoop-common:3.3.4,org.apache.hadoop:hadoop-cloud-storage:3.3.4,org.apache.hadoop:hadoop-client-runtime:3.3.4,org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.8.1,org.projectnessie.nessie-integrations:nessie-spark-extensions-3.5_2.12:0.103.2,org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.2,za.co.absa.cobrix:spark-cobol_2.12:2.8.0"
|
||||
}
|
||||
|
||||
|
||||
|
||||
conf.setAll(list(params.items()))
|
||||
|
||||
spark = SparkSession.builder.appName(workspace).config(conf=conf).getOrCreate()
|
||||
bootstrap_udfs(spark)
|
||||
|
||||
# %%
|
||||
|
||||
|
||||
|
||||
actions_audit_reader_df = spark.read.table('dremio.actionsaudit')
|
||||
actions_audit_reader_df.createOrReplaceTempView('actions_audit_reader_df')
|
||||
|
||||
# %%
|
||||
|
||||
_actions_audit_mapper_select_clause=actions_audit_reader_df.columns if False else []
|
||||
|
||||
_actions_audit_mapper_select_clause.append("DATE(action_date) AS action_date")
|
||||
|
||||
_actions_audit_mapper_select_clause.append("sub_category AS service_type")
|
||||
|
||||
_actions_audit_mapper_select_clause.append("action_count AS action_count")
|
||||
|
||||
|
||||
actions_audit_mapper_df=spark.sql(("SELECT " + ', '.join(_actions_audit_mapper_select_clause) + " FROM actions_audit_reader_df").replace("{job_id}",f"'{job_id}'"))
|
||||
actions_audit_mapper_df.createOrReplaceTempView("actions_audit_mapper_df")
|
||||
|
||||
# %%
|
||||
|
||||
print(actions_audit_mapper_df.columns)
|
||||
actions_audit_filter_df = spark.sql("select * from actions_audit_mapper_df where action_date >= COALESCE((SELECT MAX(DATE(action_date)) FROM dremio.servicemetrics), (SELECT MIN(action_date) FROM actions_audit_mapper_df))")
|
||||
actions_audit_filter_df.createOrReplaceTempView('actions_audit_filter_df')
|
||||
|
||||
# %%
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
_params = {
|
||||
"datasource": "actions_audit_filter",
|
||||
"selectFunctions" : [{'fieldName': 'service_count', 'aggregationFunction': 'SUM(action_count)'}]
|
||||
}
|
||||
|
||||
_df_flat, _grouping_specs, _rewritten_selects = preprocess_then_expand( actions_audit_filter_df,
|
||||
group_expression="action_date, service_type",
|
||||
cube="",
|
||||
rollup="",
|
||||
grouping_set="",
|
||||
select_functions=[{'fieldName': 'service_count', 'aggregationFunction': 'SUM(action_count)'}]
|
||||
)
|
||||
|
||||
_agg_exprs = [expr(f["aggregationFunction"]).alias(f["fieldName"])
|
||||
for f in _rewritten_selects
|
||||
]
|
||||
|
||||
_all_group_cols = list({c for gs in _grouping_specs for c in gs})
|
||||
|
||||
_partials = []
|
||||
for _gs in _grouping_specs:
|
||||
_gdf = _df_flat.groupBy(*_gs).agg(*_agg_exprs)
|
||||
for _col in _all_group_cols:
|
||||
if _col not in _gs:
|
||||
_gdf = _gdf.withColumn(_col, lit(None))
|
||||
_partials.append(_gdf)
|
||||
|
||||
|
||||
aggregate__3_df = reduce(lambda a, b: a.unionByName(b), _partials)
|
||||
|
||||
aggregate__3_df.createOrReplaceTempView('aggregate__3_df')
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
# %%
|
||||
|
||||
|
||||
|
||||
|
||||
_data_writer__5_fields_to_update = aggregate__3_df.columns
|
||||
_data_writer__5_set_clause=[]
|
||||
_data_writer__5_unique_key_clause= []
|
||||
|
||||
for _key in ['action_date', 'service_type']:
|
||||
_data_writer__5_unique_key_clause.append(f't.{_key} = s.{_key}')
|
||||
|
||||
for _field in _data_writer__5_fields_to_update:
|
||||
if(_field not in _data_writer__5_unique_key_clause):
|
||||
_data_writer__5_set_clause.append(f't.{_field} = s.{_field}')
|
||||
|
||||
_merge_query = '''
|
||||
MERGE INTO dremio.servicemetrics t
|
||||
USING aggregate__3_df s
|
||||
ON ''' + ' AND '.join(_data_writer__5_unique_key_clause) + ''' WHEN MATCHED THEN
|
||||
UPDATE SET ''' + ', '.join(_data_writer__5_set_clause) + ' WHEN NOT MATCHED THEN INSERT *'
|
||||
|
||||
spark.sql(_merge_query)
|
||||
|
||||
|
||||
@@ -1,197 +0,0 @@
|
||||
import marimo
|
||||
|
||||
__generated_with = "0.13.15"
|
||||
app = marimo.App()
|
||||
|
||||
|
||||
@app.cell
|
||||
def init():
|
||||
|
||||
import sys
|
||||
sys.path.append('/opt/spark/work-dir/')
|
||||
from workflow_templates.spark.udf_manager import bootstrap_udfs
|
||||
from pyspark.sql.functions import udf
|
||||
from pyspark.sql.functions import lit
|
||||
from pyspark.sql.types import StringType, IntegerType
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
from pyspark import SparkConf, Row
|
||||
from pyspark.sql import SparkSession
|
||||
import os
|
||||
import pandas as pd
|
||||
import polars as pl
|
||||
import pyarrow as pa
|
||||
from pyspark.sql.functions import expr,to_json,col,struct
|
||||
from functools import reduce
|
||||
from handle_structs_or_arrays import preprocess_then_expand
|
||||
import requests
|
||||
from jinja2 import Template
|
||||
import json
|
||||
|
||||
|
||||
from secrets_manager import SecretsManager
|
||||
|
||||
from WorkflowManager import WorkflowDSL, WorkflowManager
|
||||
from KnowledgebaseManager import KnowledgebaseManager
|
||||
from gitea_client import GiteaClient, WorkspaceVersionedContent
|
||||
|
||||
from dremio.flight.endpoint import DremioFlightEndpoint
|
||||
from dremio.flight.query import DremioFlightEndpointQuery
|
||||
|
||||
|
||||
alias_str='abcdefghijklmnopqrstuvwxyz'
|
||||
workspace = os.getenv('WORKSPACE') or 'exp360cust'
|
||||
|
||||
job_id = os.getenv("EXECUTION_ID") or str(uuid.uuid4())
|
||||
|
||||
sm = SecretsManager(os.getenv('SECRET_MANAGER_URL'), os.getenv('SECRET_MANAGER_NAMESPACE'), os.getenv('SECRET_MANAGER_ENV'), os.getenv('SECRET_MANAGER_TOKEN'))
|
||||
secrets = sm.list_secrets(workspace)
|
||||
|
||||
gitea_client=GiteaClient(os.getenv('GITEA_HOST'), os.getenv('GITEA_TOKEN'), os.getenv('GITEA_OWNER') or 'gitea_admin', os.getenv('GITEA_REPO') or 'tenant1')
|
||||
workspaceVersionedContent=WorkspaceVersionedContent(gitea_client)
|
||||
conf = SparkConf()
|
||||
params = {
|
||||
"spark.hadoop.fs.s3a.access.key": secrets.get('S3_ACCESS_KEY'),
|
||||
"spark.hadoop.fs.s3a.secret.key": secrets.get('S3_SECRET_KEY'),
|
||||
"spark.hadoop.fs.s3a.aws.region": "us-west-1",
|
||||
"spark.sql.catalog.dremio.warehouse" : secrets.get('LAKEHOUSE_BUCKET'),
|
||||
"spark.sql.catalog.dremio" : "org.apache.iceberg.spark.SparkCatalog",
|
||||
"spark.sql.catalog.dremio.type" : "hadoop",
|
||||
"spark.hadoop.fs.s3a.impl": "org.apache.hadoop.fs.s3a.S3AFileSystem",
|
||||
"spark.hadoop.fs.s3.impl": "org.apache.hadoop.fs.s3a.S3AFileSystem",
|
||||
"spark.hadoop.fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem",
|
||||
"spark.sql.extensions": "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions",
|
||||
"spark.jars.packages": "com.amazonaws:aws-java-sdk-bundle:1.12.262,com.github.ben-manes.caffeine:caffeine:3.2.0,org.apache.iceberg:iceberg-aws-bundle:1.8.1,org.apache.iceberg:iceberg-common:1.8.1,org.apache.iceberg:iceberg-core:1.8.1,org.apache.iceberg:iceberg-spark:1.8.1,org.apache.hadoop:hadoop-aws:3.3.4,com.amazonaws:aws-java-sdk-bundle:1.11.901,org.apache.hadoop:hadoop-common:3.3.4,org.apache.hadoop:hadoop-cloud-storage:3.3.4,org.apache.hadoop:hadoop-client-runtime:3.3.4,org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.8.1,org.projectnessie.nessie-integrations:nessie-spark-extensions-3.5_2.12:0.103.2,org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.2,za.co.absa.cobrix:spark-cobol_2.12:2.8.0"
|
||||
}
|
||||
|
||||
|
||||
|
||||
conf.setAll(list(params.items()))
|
||||
|
||||
spark = SparkSession.builder.appName(workspace).config(conf=conf).getOrCreate()
|
||||
bootstrap_udfs(spark)
|
||||
return expr, job_id, lit, preprocess_then_expand, reduce, spark
|
||||
|
||||
|
||||
@app.cell
|
||||
def actions_audit_reader(spark):
|
||||
|
||||
|
||||
|
||||
actions_audit_reader_df = spark.read.table('dremio.actionsaudit')
|
||||
actions_audit_reader_df.createOrReplaceTempView('actions_audit_reader_df')
|
||||
return (actions_audit_reader_df,)
|
||||
|
||||
|
||||
@app.cell
|
||||
def actions_audit_mapper(actions_audit_reader_df, job_id, spark):
|
||||
|
||||
_actions_audit_mapper_select_clause=actions_audit_reader_df.columns if False else []
|
||||
|
||||
_actions_audit_mapper_select_clause.append("DATE(action_date) AS action_date")
|
||||
|
||||
_actions_audit_mapper_select_clause.append("sub_category AS service_type")
|
||||
|
||||
_actions_audit_mapper_select_clause.append("action_count AS action_count")
|
||||
|
||||
|
||||
actions_audit_mapper_df=spark.sql(("SELECT " + ', '.join(_actions_audit_mapper_select_clause) + " FROM actions_audit_reader_df").replace("{job_id}",f"'{job_id}'"))
|
||||
actions_audit_mapper_df.createOrReplaceTempView("actions_audit_mapper_df")
|
||||
return (actions_audit_mapper_df,)
|
||||
|
||||
|
||||
@app.cell
|
||||
def actions_audit_filter(actions_audit_mapper_df, spark):
|
||||
|
||||
print(actions_audit_mapper_df.columns)
|
||||
actions_audit_filter_df = spark.sql("select * from actions_audit_mapper_df where action_date >= COALESCE((SELECT MAX(DATE(action_date)) FROM dremio.servicemetrics), (SELECT MIN(action_date) FROM actions_audit_mapper_df))")
|
||||
actions_audit_filter_df.createOrReplaceTempView('actions_audit_filter_df')
|
||||
return (actions_audit_filter_df,)
|
||||
|
||||
|
||||
@app.cell
|
||||
def aggregate__3(
|
||||
actions_audit_filter_df,
|
||||
expr,
|
||||
lit,
|
||||
preprocess_then_expand,
|
||||
reduce,
|
||||
):
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
_params = {
|
||||
"datasource": "actions_audit_filter",
|
||||
"selectFunctions" : [{'fieldName': 'service_count', 'aggregationFunction': 'SUM(action_count)'}]
|
||||
}
|
||||
|
||||
_df_flat, _grouping_specs, _rewritten_selects = preprocess_then_expand( actions_audit_filter_df,
|
||||
group_expression="action_date, service_type",
|
||||
cube="",
|
||||
rollup="",
|
||||
grouping_set="",
|
||||
select_functions=[{'fieldName': 'service_count', 'aggregationFunction': 'SUM(action_count)'}]
|
||||
)
|
||||
|
||||
_agg_exprs = [expr(f["aggregationFunction"]).alias(f["fieldName"])
|
||||
for f in _rewritten_selects
|
||||
]
|
||||
|
||||
_all_group_cols = list({c for gs in _grouping_specs for c in gs})
|
||||
|
||||
_partials = []
|
||||
for _gs in _grouping_specs:
|
||||
_gdf = _df_flat.groupBy(*_gs).agg(*_agg_exprs)
|
||||
for _col in _all_group_cols:
|
||||
if _col not in _gs:
|
||||
_gdf = _gdf.withColumn(_col, lit(None))
|
||||
_partials.append(_gdf)
|
||||
|
||||
|
||||
aggregate__3_df = reduce(lambda a, b: a.unionByName(b), _partials)
|
||||
|
||||
aggregate__3_df.createOrReplaceTempView('aggregate__3_df')
|
||||
|
||||
|
||||
|
||||
|
||||
return (aggregate__3_df,)
|
||||
|
||||
|
||||
@app.cell
|
||||
def data_writer__5(aggregate__3_df, spark):
|
||||
|
||||
|
||||
|
||||
|
||||
_data_writer__5_fields_to_update = aggregate__3_df.columns
|
||||
_data_writer__5_set_clause=[]
|
||||
_data_writer__5_unique_key_clause= []
|
||||
|
||||
for _key in ['action_date', 'service_type']:
|
||||
_data_writer__5_unique_key_clause.append(f't.{_key} = s.{_key}')
|
||||
|
||||
for _field in _data_writer__5_fields_to_update:
|
||||
if(_field not in _data_writer__5_unique_key_clause):
|
||||
_data_writer__5_set_clause.append(f't.{_field} = s.{_field}')
|
||||
|
||||
_merge_query = '''
|
||||
MERGE INTO dremio.servicemetrics t
|
||||
USING aggregate__3_df s
|
||||
ON ''' + ' AND '.join(_data_writer__5_unique_key_clause) + ''' WHEN MATCHED THEN
|
||||
UPDATE SET ''' + ', '.join(_data_writer__5_set_clause) + ' WHEN NOT MATCHED THEN INSERT *'
|
||||
|
||||
spark.sql(_merge_query)
|
||||
|
||||
|
||||
return
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
app.run()
|
||||
File diff suppressed because one or more lines are too long
Reference in New Issue
Block a user