About

This program take a DAG input as json and converts it to Airflow file.

Run

mvn clean install

How to use

// Read file
String fileName = "graph.json";
ClassLoader classLoader = getClass().getClassLoader();
File file = new File(classLoader.getResource(fileName).getFile());
String content = IOUtils.toString(file.toURI(), Charset.defaultCharset());
assertNotNull(content);

// Read Graph from file
ObjectMapper objectMapper = new ObjectMapper();
objectMapper.setPropertyNamingStrategy(PropertyNamingStrategy.SNAKE_CASE);
Graph graph = objectMapper.readValue(content, Graph.class);
assertNotNull(graph);

// Setup graph
GraphInstance graphInstance = new GraphInstance();
graphInstance.initGraph(graph);

List<String> result = graphInstance.travers(new AirflowTraversal());
assertNotNull(result);
assertTrue(result.size() > 0);

result.forEach(System.out::println);

Output

end.set_upstream([sink_2__needs__transformation_2, sink_1__needs__transformation_1])
sink_2__needs__transformation_2.set_upstream(transformation_2__needs_source_read_from_redshift_2_and_3)
transformation_2__needs_source_read_from_redshift_2_and_3.set_upstream([source_read_from_redshift_2, source_read_from_redshift_3])
source_read_from_redshift_2.set_upstream(start)
source_read_from_redshift_3.set_upstream(start)
sink_1__needs__transformation_1.set_upstream(transformation_1__needs_source_read_from_redshift_1)
transformation_1__needs_source_read_from_redshift_1.set_upstream(source_read_from_redshift_1)
source_read_from_redshift_1.set_upstream(start)

It will generate out.py - you can put this file to airflow Dag folder to execute.

Graph Json

The file /src/main/resources/graph.json contains the json of this graph. You can edit to change it.

{
  "start": {
    "id": "start",
    "type": "no_op"
  },
  "source": [
    {
      "id": "source_read_from_redshift_1",
      "type": "read_from_redshift",
      "depends_on": [
        "start"
      ],
      "config": {
        "redshift": {
          "host": "redshift.internal",
          "query": "select * from table_user"
        }
      }
    },
    {
      "id": "source_read_from_redshift_2",
      "type": "read_from_redshift",
      "depends_on": [
        "start"
      ],
      "config": {
        "redshift": {
          "host": "redshift.internal",
          "query": "select * from table_order"
        }
      }
    },
    {
      "id": "source_read_from_redshift_3",
      "type": "read_from_redshift",
      "depends_on": [
        "start"
      ],
      "config": {
        "redshift": {
          "host": "redshift.internal",
          "query": "select * from table_card"
        }
      }
    }
  ],
  "transformation": [
    {
      "id": "transformation_1__needs_source_read_from_redshift_1",
      "depends_on": [
        "source_read_from_redshift_1"
      ],
      "type": "docker_execute",
      "config": {
        "docker": {
          "image": "custom_image:1.0.0",
          "volumn": "${b4b0535a-f87d-49fd-885f-7e237e557cf4}.output.s3"
        }
      }
    },
    {
      "id": "transformation_2__needs_source_read_from_redshift_2_and_3",
      "depends_on": [
        "source_read_from_redshift_2",
        "source_read_from_redshift_3"
      ],
      "type": "spark_execute",
      "config": {
        "spark": {
          "program": "s3://.../program.jar"
        }
      }
    }
  ],
  "sink": [
    {
      "id": "sink_1__needs__transformation_1",
      "depends_on": [
        "transformation_1__needs_source_read_from_redshift_1"
      ],
      "type": "map_s3_to_athena",
      "config": {
        "athena": {
          "input_bucket": "${5fc684e4-cfd1-44f0-842b-f62fa9163327}.output.s3",
          "column_defination": "id int, ...",
          "table_name": "table_1234"
        }
      }
    },
    {
      "id": "sink_2__needs__transformation_2",
      "depends_on": [
        "transformation_2__needs_source_read_from_redshift_2_and_3"
      ],
      "type": "map_s3_to_athena",
      "config": {
        "athena": {
          "input_bucket": "${de2f0dfd-ea15-4220-8b6b-0b48a011c29c}.output.s3",
          "column_defination": "id int, ...",
          "table_name": "table_1234_abcd"
        }
      }
    }
  ],
  "end": {
    "id": "end",
    "depends_on": [
      "sink_1__needs__transformation_1",
      "sink_2__needs__transformation_2"
    ],
    "type": "docker_execute",
    "config": {
      "docker": {
        "image": "custom_image:1.0.0"
      }
    }
  }
}