A Beginner’s Guide to Airflow

Airflow is used to create code pipeline where we can schedule and monitor our workflows. A workflow can be a collection of tasks to be executed like a flowchart.

It is like an orchestra conductor that controls all different data processing tools/tasks under one roof.

Why Airflow?

  • Open source.
  • Can handle upstream/downstream in an elegant way.
  • Jobs can pass parameters to other jobs downstream.
  • Ease of deployment of workflow.
  • Easy to reprocess historical jobs by date or rerun for specific intervals.
  • Integration with a lot of infrastructure.
  • Job testing through airflow itself.
  • Accessibility of log files and other meta-data through the web GUI.
  • Data sensors to trigger Directed Acyclic Graph (DAG) when data arrives.

Airflow Architecture:

Metadata: It stores the state of tasks and workflow.
Scheduler: It is a user DAG, with the state of tasks in metadata database. It decides what needs to execute.
Executor: It is a message queuing process. Decides which worker will execute each task.

Setting up airflow:

Use the following commands to install airflow

$ sudo pip install apache-airflow

You can install extra features like

$ sudo pip install apache-airflow[postgres,s3]

Airflow requires database to be initiated before you run tasks

$ airflow initdb

After installation the folder structure will be something like this.

If you don’t see ‘dags’ folder then you can create by yourself and name it ‘dags’. You shall dump all your python task files in that folder.

To start the web server

$ Airflow webserver -p 8080

Goto https://localhost:8080 to see airflow GUI. You should see something like this:

Airflow relies on 4 cores:

  1. DAG.
  2. Tasks.
  3. Scheduler.
  4. X-com.

Directed Acyclic Graph (DAG):

It is a collection of all tasks which you want to run in an organized way that shows their relationships and dependencies.

Basically a DAG is just a python file, which is used to organise tasks and set their execution content. DAGs do not perform any actual computation. Basically a DAG task contains 2 type of things:

  1. Operators
    It determines what actually gets done. It triggers a certain action like running a bash command, executing a python function, etc.
    There are different kinds of operators like:
    Bash operator: Executes a bash command.
    Python operator: Calls an arbitrary python function.
    Hive operator: Executes hql code or hive script in specific hive database.
    Bigquery operator: Executes Google BigQuery SQL query in a specific BigQuery database.
  2. Sensors : A certain type of operator that will keep running until a certain criteria is met.

Tasks :
Tasks are the elements that actually “do work” which we want to be performed. It is our job to write the configuration and organise tasks in specific order to create data pipeline.

Once the operator is instantiated, it is referred to as a “task”. An operator describes single task in a workflow

Scheduler :
It monitors all tasks and all dags. And trigger the task instance whose dependencies have met.
We create new python file in dags folder
Importing packages in our python file:

import datetime as dt
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator

Next setting default arguments

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': dt.datetime(2019, 6, 17, 8, 40, 00),
    'retries': 1,
}

Depends_on_past: we will be in a situation where  a task t1 is dependent on its previous execution. In that case we will use this.

Start_date: starting date of execution
Defining our tasks, dags, functions

def greet():
    print("Hello how are you ?")
    return 'greeted'

def response():
    return "im fine what about you?"

with DAG('YourDagName', default_args=default_args, schedule_interval='@daily',) as dag:
    opr_hello = BashOperator(task_id='say_Hi',bash_command='echo "Hi!!"')
    opr_greet = PythonOperator(task_id='greet', python_callable=greet)
    opr_sleep = BashOperator(task_id='sleep_me', bash_command='sleep 2')
    opr_respond = PythonOperator(task_id='respond', python_callable=response)

In this, I am using 2 simple functions and with dag passing my default arguments and also schedule time of execution.

After that we created 4 tasks in which 2 are python and other 2 are bash. If we want to execute a python function we need to use python operator and also the same with the bash operator. Python operator will take at least 2 arguments which is task id and python callable and for bash, task id and bash command.

Defining task dependencies or order in which the tasks should be executed in.

Task dependencies are set using

  • The set_upstream and set_downstream operator.
  • The bitshift operator <<  and >>.

t1.set_downstream(t2)
t1>>t2

The above 2 lines mean the same. This means t2 will depend on t1 to run successfully.

opr_hello >> opr_greet >> opr_sleep >> opr_respond

If you wrap up the whole code in one file and put that file in dags folder then you can see your dag in Airflow GUI. If you click on your DAG you will see tree view and many other options to view your data in different forms.

Dagrun are dags that runs at certain time.
TaskInstances are task belongs to dagrun.
Each dagrun and task instance is associated with an entry in airflow’s metadata database that logs their state(eg: queued, running, failed, skipped, etc).
X-com: XCom which means cross communication allow airflow tasks to send and receive data/status. XCom data are stored in the airflow database in the form of key-value pairs.

That’s all guys. Thanks for the read!

This story is co-authored by Santosh Kumar and PV Subbareddy. Santosh is a Software Engineer specializing on Cloud Services and DevOps. Subbareddy is a Big Data Engineer specializing on AWS Big Data Services and Apache Spark Ecosystem.

Building a Simple React Application using Redux

In this article I will be building a simple react app that uses redux to manage its state. If you are a complete beginner, I recommend first reading my A Beginners Guide to Understanding Redux article then following this article. For this article, I have built a simple react app that renders a dummy signup form and a table which displays the list of employee details (dummy data). We shall integrate it with redux but first do the initial setup to follow along.

Initial Setup

Please clone this specific branch ‘without-redux’ from my GitHub repository.

git clone -b without-redux --single-branch https://github.com/koushik-bitzop/newempdetails.git

Run the above command on your terminal, this should create a folder and clone all the required code files from my repository.
To get started install the dependencies:

sudo npm install

Start the application server:

sudo npm start

You should see the app running on your localhost, like this:

The code files that you have cloned and the below are the same.

App.js

import React, {Component} from 'react';
import Header from './components/Header';
import Routes from './components/Routes';
import { BrowserRouter as Router} from 'react-router-dom';

class App extends Component {
 state = {  }
 render() {
   return (
   <Router>
       <Header/>
       <div className="row">
           <Routes />
       </div>
   </Router>
   );
 }
}

export default App;

Routes.js

import React, {Component} from 'react';
import {Switch} from 'react-router';
import {Route} from 'react-router-dom';
import FormdataTable from './FormdataTable';
import SignupForm from './SignupForm';

class Routes extends Component {
   render() {
       return ( 
           <Switch>
               <Route path={'/'} exact component={SignupForm } ></Route>
               <Route path={'/viewlist'} exact component={FormdataTable}></Route>
           </Switch>
       );
   }
}

export default Routes;

Header.js

import React,{Component} from 'react';
import {Link, NavLink} from 'react-router-dom';

class Header extends Component{
   render(){
       return(
           <nav className="navbar navbar-expand-lg navbar-light bg-light">
                  
               <Link to="/" className="navbar-brand" href="#"><h1>Wisdatum</h1></Link>

               <div className="collapse navbar-collapse" id="navbarNav">
                   <ul className="navbar-nav">
                       <li className="nav-item active">
                           <NavLink to="/" className="nav-link" href="#">Add new </NavLink>
                       </li>
                       <li className="nav-item">
                           <NavLink to="/viewlist" className="nav-link" href="#">View list</NavLink>
                       </li>
                   </ul>
               </div>

           </nav>
       );
   }
}

export default Header;

SignupForm.js

import React, {Component} from 'react';
import {Form, FormGroup, FormControl, FormLabel, Button} from 'react-bootstrap';

class SignupForm extends Component {
   constructor(props) {
   super(props);
   this.state = {
       firstname: '',
       lastname: '',
       email: '',
       mobile: '',
       city: ''
   };

   // This binding is necessary to make `this` work in the callback
   this.handleSubmit = this.handleSubmit.bind(this);
   }
handleSubmit(){
   let formdata = {
       firstname: this.state.firstname,
       lastname: this.state.lastname,
       email: this.state.email,
       mobile: this.state.mobile,
       city: this.state.city
   };
   console.log("submitted",formdata);
   this.setState({
       firstname: '',
       lastname: '',
       email: '',
       mobile: '',
       city: ''
   });
}
   render() {
       return (
           <div className="col-md-4 offset-md-4">
               <Form>
                   <h2 style={{"textAlign":"center", "marginTop":"20px"}}>Enter Employee Details</h2>
                   <hr/>
                   <FormGroup>
                       <FormLabel>Firstname</FormLabel>
                       <FormControl
                           type="text"
                           name="firstname"
                           placeholder="Firstname"
                           onChange={e => {
                               this.setState({[e.target.name]:e.target.value});
                           }}
                           value = {this.state.firstname}
                       />
                   </FormGroup>
              
                   <FormGroup>
                       <FormLabel>Lastname</FormLabel>
                       <FormControl
                           type="text"
                           name="lastname"
                           placeholder="Lastname"
                           onChange={e => {
                               this.setState({[e.target.name]:e.target.value});
                           }}
                           value = {this.state.lastname}
                       />
                   </FormGroup>

                   <FormGroup>
                       <FormLabel>Email</FormLabel>
                       <FormControl
                           type="text"
                           name="email"
                           placeholder="Email"
                           onChange={e => {
                               this.setState({[e.target.name]:e.target.value});
                           }}
                           value = {this.state.email}
                       />
                   </FormGroup>

                   <FormGroup>
                       <FormLabel>Mobile</FormLabel>
                       <FormControl
                           type="text"
                           name="mobile"
                           placeholder="Mobile"
                           onChange={e => {
                               this.setState({[e.target.name]:e.target.value});
                           }}
                           value = {this.state.mobile}
                       />
                   </FormGroup>

                   <FormGroup>
                       <FormLabel>City</FormLabel>
                       <FormControl
                           type="text"
                           name="city"
                           placeholder="City/Village"
                           onChange={e => {
                               this.setState({[e.target.name]:e.target.value});
                           }}
                           value = {this.state.city}
                   />
                   </FormGroup>
                   <Button onClick={this.handleSubmit}>Submit</Button>
               </Form>
           </div>
        );
   }
}
export default SignupForm;

FormdataTable.js

import React,{Component} from 'react';
import empdata from '../data/employeedata.json';

class FormdataTable extends Component {
   state = {  }
   render() {
       console.log(empdata);
       return (
           <div className="col-md-10 offset-md-1">
               <h2 style={{"textAlign":"center", "marginTop":"20px", "marginBottom":"20px"}}>New Employee Details</h2>
               <table className="table table-hover">
                   <thead>
                       <tr>
                           <th scope="col">#</th>
                           <th scope="col">Fist Name</th>
                           <th scope="col">Last Name</th>
                           <th scope="col">Email</th>
                           <th scope="col">Mobile</th>
                           <th scope="col">City</th>
                           <th scope="col">Options</th>
                       </tr>
                   </thead>
                   <tbody>
                       {
                       empdata.map((emp, index) => {
                       return(
                           <tr key={index}>
                                   <td>{index+1}</td>
                                   <td>{emp.firstname}</td>
                                   <td>{emp.lastname}</td>
                                   <td>{emp.email}</td>
                                   <td>{emp.mobile}</td>
                                   <td>{emp.city}</td>
                                   <td>
                                       <button
                                           type="button"
                                           onClick={()=>this.handleEdit(index)}
                                           className="btn btn-sm btn-primary">Edit
                                       </button>
                                       {" | "}
                                       <button
                                           type="button"
                                           onClick={()=>this.handleDelete(index)}
                                           className="btn btn-sm btn-danger">Delete
                                       </button>
                                   </td>
                           </tr>
                       )})
                       }
                   </tbody>
               </table>
           </div>
       );
   }
}
export default FormdataTable;

Before we begin integrating with redux, please review and understand the code above. I have used bootstrap for styling and react-router for creating the navigation/routes in the application.

Integrating with redux:

The above diagram depicts redux workflow in a nutshell, UI triggers actions like form-submit, edit & delete in table. Each action represents an action object, these action objects are then dispatched to Reducer function which updates the Store. Store contains the State. All the components get their data from the store and will be updated with new state. This new state will re-render and defines the new UI changes, repeats the cycle.

Redux is external to react we shall need redux & react-redux library to connect to redux with react.

npm install redux react-redux --save

We shall go step by step.

Step1: Create a root reducer function

Reducer function always talks to the store, updates the store and creates the store.

allreducer.js (create in src folder)

import empdata_json from '../data/employeedata.json';

function empdata(state = empdata_json, action){
    switch(action.type){
        default:
            return state;
    }
}

export default empdata;

The above reducer function accepts two arguments current state and an action object. Initial values of this state is set to dummy data imported from employeedata.json file. Whenever this function is invoked, based on the action type, the corresponding block is executed in the switch. Though the logic of the blocks may differ, they all pretty much do the same. Take the payload and return new state to update the store.

Step2: Create the store, make it available to whole tree

We have to import the above reducer function and use createStore method available in redux to create the store.

To make this store available to the whole app component tree we wrap the whole nested app component tree in Provider component available in react-redux store as props.

The <Provider /> makes the Redux store available to any nested components that have been wrapped in the connect() function. We will learn about connect() in the next step. Since any React component in a React Redux app can be connected, most applications will render a <Provider> at the top level, with the entire app’s component tree inside of it.

Normally, you can’t use a connected component unless it is nested inside of a <Provider>.

index.js (modify this file)

import React from 'react';
import ReactDOM from 'react-dom';
import './index.css';
import App from './App';
import * as serviceWorker from './serviceWorker';
import empdatareducer from '../src/allreducer';
import {createStore} from 'redux';
import {Provider} from 'react-redux';

const store = createStore(empdatareducer);
ReactDOM.render(<Provider store={store}><App /></Provider>, document.getElementById('root'));

// If you want your app to work offline and load faster, you can change
// unregister() to register() below. Note this comes with some pitfalls.
// Learn more about service workers: https://bit.ly/CRA-PWA
serviceWorker.unregister();

Step3: Connecting with redux

Now the store is available to whole component tree, components have to be connected to redux and be mapped with the store to get data(as props). The react-redux library comes with a function called connect, which is how you can feed data from Redux’s store into your React components and also dispatch actions to reducer function. The connect function is commonly passed 1 or 2 arguments:

First, a mapStateToProps function that takes out pieces of state out of Redux and assigns them to props that your React component will use.

And often a second argument: a mapDispatchToProps function which binds action creator functions(return action object) with dispatch, So you can just write props.actionName() and you don’t have to write dispatch: actionObject, for every event or action. The dispatch is invoked as soon as action creator returns an action object.

Remember, you can’t dispatch an action to a specific reducer function, Whenever there is a dispatch, it is received by all reducers, only those blocks where action type is matched are executed.

As our FormdataTable.js component requires the store data, we map its props with store.

FormdataTable.js (modify this file)

import React,{Component} from 'react';
import {connect } from 'react-redux';

class FormdataTable extends Component {
  state = {  }
  render() {
      //console.log(this.props.empdata);
      return (
          <div className="col-md-10 offset-md-1">
              <h2 style={{"textAlign":"center", "marginTop":"20px", "marginBottom":"20px"}}>New Employee Details</h2>
              <table className="table table-hover">
                  <thead>
                      <tr>
                          <th scope="col">#</th>
                          <th scope="col">Fist Name</th>
                          <th scope="col">Last Name</th>
                          <th scope="col">Email</th>
                          <th scope="col">Mobile</th>
                          <th scope="col">City</th>
                          <th scope="col">Options</th>
                      </tr>
                  </thead>
                  <tbody>
                      {
                      this.props.empdata.map((emp, index) => {
                      return(
                          <tr key={index}>
                                  <td>{index+1}</td>
                                  <td>{emp.firstname}</td>
                                  <td>{emp.lastname}</td>
                                  <td>{emp.email}</td>
                                  <td>{emp.mobile}</td>
                                  <td>{emp.city}</td>
                                  <td>
                                      <button
                                          type="button"
                                          onClick={()=>this.handleEdit(index)}
                                          className="btn btn-sm btn-primary">Edit
                                      </button>
                                      {" | "}
                                      <button
                                          type="button"
                                          onClick={()=>this.handleDelete(index)}
                                          className="btn btn-sm btn-danger">Delete
                                      </button>
                                  </td>
                          </tr>
                      )})
                      }
                  </tbody>
              </table>
          </div>
      );
  }
}
function mapStateToProps(state){
     return {
       empdata : state
   };
}
export default connect(mapStateToProps,null)(FormdataTable);

Our signup formdata is to be updated with the store, this is an action and should be dispatched. We will first write an action creator function and then bind it with dispatch using bindActionCreators available in redux. Only then we could dispatch it to the reducer to update with the store.

allactions.js (create in src folder)

export function addNewEmp(payload){
   const action = {
       type: "ADD_NEW_EMP",
       payload
   }
   return action;
}

In the above case payload is our formdata. Let us bind this action creator with dispatch and connect to redux to dispatch our action object containing payload(formdata).

SignupForm.js (modify this file)

import React,{Component} from 'react';
import {Form, FormGroup, FormControl, FormLabel, Button} from 'react-bootstrap';

import {connect} from 'react-redux';
import {bindActionCreators} from 'redux';
import {addNewEmp} from '../allactions';

class SignupForm extends Component {
  constructor(props) {
  super(props);
  this.state = {
      firstname: '',
      lastname: '',
      email: '',
      mobile: '',
      city: ''
  };

  // This binding is necessary to make `this` work in the callback
  this.handleSubmit = this.handleSubmit.bind(this);
  }
handleSubmit(){
  let formdata = {
      firstname: this.state.firstname,
      lastname: this.state.lastname,
      email: this.state.email,
      mobile: this.state.mobile,
      city: this.state.city
  };
  this.props.addNewEmp(formdata);
  console.log("submitted",formdata);
  this.setState({
      firstname: '',
      lastname: '',
      email: '',
      mobile: '',
      city: ''
  });
}
  render() {
      return (
          <div className="col-md-4 offset-md-4">
              <Form>
                  <h2 style={{"textAlign":"center", "marginTop":"20px"}}>Enter Employee Details</h2>
                  <hr/>
                  <FormGroup>
                      <FormLabel>Firstname</FormLabel>
                      <FormControl
                          type="text"
                          name="firstname"
                          placeholder="Firstname"
                          onChange={e => {
                              this.setState({[e.target.name]:e.target.value});
                          }}
                          value = {this.state.firstname}
                      />
                  </FormGroup>
            
                  <FormGroup>
                      <FormLabel>Lastname</FormLabel>
                      <FormControl
                          type="text"
                          name="lastname"
                          placeholder="Lastname"
                          onChange={e => {
                              this.setState({[e.target.name]:e.target.value});
                          }}
                          value = {this.state.lastname}
                      />
                  </FormGroup>

                  <FormGroup>
                      <FormLabel>Email</FormLabel>
                      <FormControl
                          type="text"
                          name="email"
                          placeholder="Email"
                          onChange={e => {
                              this.setState({[e.target.name]:e.target.value});
                          }}
                          value = {this.state.email}
                      />
                  </FormGroup>

                  <FormGroup>
                      <FormLabel>Mobile</FormLabel>
                      <FormControl
                          type="text"
                          name="mobile"
                          placeholder="Mobile"
                          onChange={e => {
                              this.setState({[e.target.name]:e.target.value});
                          }}
                          value = {this.state.mobile}
                      />
                  </FormGroup>

                  <FormGroup>
                      <FormLabel>City</FormLabel>
                      <FormControl
                          type="text"
                          name="city"
                          placeholder="City/Village"
                          onChange={e => {
                              this.setState({[e.target.name]:e.target.value});
                          }}
                          value = {this.state.city}
                  />
                  </FormGroup>
                  <Button onClick={this.handleSubmit}>Submit</Button>
              </Form>
          </div>
       );
  }
}

function mapDispatchToProps(dispatch){
   return bindActionCreators({addNewEmp},dispatch)
}
export default connect (null, mapDispatchToProps)(SignupForm)

That simple, we have successfully managed state with redux, now let’s test it.

Now, please try to implement the delete functionality by yourself like shown below

Before deletion:

After deletion:

If you could do it you are doing great. If not please feel free to refer my code files below.

https://github.com/koushik-bitzop/newempdetails/commit/cd9712a89c4de7625cecbd6e494199f0ad5dd20f?diff=split

I hope this article was helpful to you in learning redux. Thanks for the read!

This story is authored by Koushik Busim. Koushik is a software engineer and a keen data science and machine learning enthusiast.