mesmacosta

from datetime import datetime, timedelta

from airflow.models import DAG
from airflow.operators.python import PythonOperator
from airflow.utils.session import provide_session

from alvin_integration.producers.airflow.pipeline.extractor import extract_dag_metadata

seven_days_ago = datetime.combine(
    datetime.today() - timedelta(1), datetime.min.time()

mesmacosta

  "facets": {
    "alvin": {
      "connection_id": "google_cloud_default",
      "execution": {
        "alvin_package_version": "0.15.7",
        "platform_id": "airflowprd",
        "duration": 41.69428,
        ...
      },
      "job_id": "airflow_2141512512_fcf9812412412412412421412"

mesmacosta

{
  "eventType": "START",
  "eventTime": "2020-12-09T23:37:31.081Z",
  "run": {
    "runId": "3b452093-782c-4ef2-9c0c-aafe2aa6f34d",
  },
  "job": {
    "namespace": "my-scheduler-namespace",
    "name": "myjob.mytask",
    "facets": {...}

mesmacosta

    def execute(self, context: 'Context'):
# ... code suppresed for readability
        if isinstance(self.sql, str):
            job_id: Union[str, List[str]] = self.hook.run_query(
                sql=self.sql,
                destination_dataset_table=self.destination_dataset_table,
                write_disposition=self.write_disposition,
                allow_large_results=self.allow_large_results,
                flatten_results=self.flatten_results,
                udf_config=self.udf_config,

mesmacosta

def execute(self, context: Any):
        hook = BigQueryHook(
            gcp_conn_id=self.gcp_conn_id,
            delegate_to=self.delegate_to,
            impersonation_chain=self.impersonation_chain,
        )
        self.hook = hook

        job_id = self._job_id(context)

mesmacosta

handler = logging.StreamHandler()

# Set debug level for the client
es_logger.setLevel(logging.DEBUG)
es_logger.addHandler(handler)
es_logger.disabled = False

# Set trace logger
es_trace_logger = logging.getLogger('elasticsearch.trace')
es_trace_logger.setLevel(logging.DEBUG)

mesmacosta

sql_objects:
  - name: 'functions'
    enabled: True

mesmacosta

SELECT f.SCHEMA_NAME AS SCHEMA_NAME,
       f.FUNCTION_NAME AS function_name,
       f.INPUT_PARAMETER_COUNT AS input_parameter_count,
       f.RETURN_VALUE_COUNT AS return_value_count,
       f.DEFINITION AS definition,
       f.IS_VALID AS is_valid,
       f.OWNER_NAME AS owner_name,
       f.CREATE_TIME AS create_time
FROM SYS.FUNCTIONS f
// ... Omitted to improve readability

mesmacosta

{
  "key": "functions",
  "type": "function",
  "name": "function_name",
  "fields": [
    {
      "source": "schema_name",
      "target": {
        "field_name": "schema_name",
        "model": "tag",

mesmacosta

CREATE TABLE credit_tab_pii (Name varchar(20), CREDIT_CARD varchar(19)) 
WITH MASK (CREDIT_CARD USING credit_mask(credit_card));