Stream

September 2018 Update: Ansible 2.7 (to be released around October 2018) will include a new reboot module, which makes reboots a heck of a lot simpler (whether managing Windows, Mac, or Linux!):

- name: Reboot the server and wait for it to come back up.
  reboot:

That's it! Much easier than the older technique I used in Ansible < 2.7!

One pattern I often need to implement in my Ansible playbooks is "configure-reboot-configure", where you change some setting that requires a reboot to take effect, and you have to wait for the reboot to take place before continuing on with the rest of the playbook run.

For example, on my Raspberry Pi Dramble project, before installing Docker and Kubernetes, I need to make sure the Raspberry Pi's /boot/cmdline.txt file contains a couple cgroup features so Kubernetes runs correctly. But after adding these options, I also have to reboot the Pi.

If you just add a task like command: reboot (run with become/sudo), you might run into this annoying error during a playbook run:

When that happens, the playbook fails and you have to run it again to get back to the point where you can start configuring things again.

Luckily, Ansible has a way of dealing with this special case where you are basically dropping the ability to connect to the server Ansible is managing, at least temporarily, using async and the special wait_for_connection module.

Here's how I rewrote my tasks so they would correctly reboot the Raspberry Pi, then wait for it to be available again before proceeding with the rest of the playbook:

Notes

1. I don't want to reboot the Pi every time my playbook runs, but instead only when it needs to reboot. So I registered task_result and used task_result is changed as a condition for the reboot. This way, the second time I run my playbook I don't have to sit around and wait for an unnecessary reboot!
2. I used the shell module instead of command for the reboot task, because shell supports builtins like &&. And I used sleep 5 && reboot instead of just reboot, because Ansible needs a couple seconds to wrap up it's connection so the task can complete. If you just do a reboot or shutdown -r now, then the server will drop the SSH connection before Ansible can close it cleanly, and you'll get the same error I mentioned earlier.
3. You may need different values for your wait_for_connection task; my Raspberry Pis usually reboot within a minute or so; but some cloud servers and other types of servers could take much longer.

Running with ansible as an ad-hoc command

Sometimes I also need to do the same thing using a quick ad-hoc command. This is pretty easy to translate using the -B [seconds] ('background' option, with timeout in seconds) and -P [seconds] ('polling' option, with delay in seconds; 0 to disable polling):

ansible all -i inventory -b -B 1 -P 0 -m shell -a "sleep 5 && reboot"

Ansible is popular for its simplicity of installation, ease of use in what concerns the connectivity to clients, its lack of agent for Ansible clients and the multitude of skills. It functions by connecting via SSH to the clients, so it doesn't need a special agent on the client-side, and by pushing modules to the clients, the modules are then executed locally on the client-side and the output is pushed back to the Ansible server.

Since it uses SSH, it can very easily connect to clients using SSH-Keys, simplifying though the whole process. Client details, like hostnames or IP addresses and SSH ports, are stored in files called inventory files. Once you have created an inventory file and populated it, ansible can use it.

In this Ansible tutorial for beginners, you will learn Ansible step by step:

• What is Ansible?
• Why use Ansible?
• History of Ansible
• Important terms used in Ansible
• Ansible Installation in Linux
• Ansible ad-hoc commands
• Ansible Playbooks
• Ansible Roles
• Ansible Case Study
• Ansible Commands Cheat Sheet

Why use Ansible?

Here are some important pros/benefits of using Ansible

• One of the most significant advantages of Ansible is that it is free to use by everyone.
• It does not need any special system administrator skills to install and use Ansible, and the official documentation is very comprehensive.
• Its modularity regarding plugins, modules, inventories, and playbooks make Ansible the perfect companion to orchestrate large environments.
• Ansible is very lightweight and consistent, and no constraints regarding the operating system or underlying hardware are present.
• It is also very secure due to its agentless capabilities and due to the use of OpenSSH security features.
• Another advantage that encourages the adoption of Ansible is its smooth learning curve determined by the comprehensive documentation and easy to learn structure and configuration.

History of Ansible

Here, are important land marks from the history of ansible:

• In February 2012 the Ansible project began. It was first developed by Michael DeHaan, the creator of Cobbler and Func, Fedora Unified Network Controller.
• Initially called AnsibleWorks Inc, the company funding the ansible tool was acquired in 2015 by RedHat and later on, along with RedHat, moved under the umbrella of IBM.
• In the present, Ansible comes included in distributions like Fedora Linux, RHEL, Centos and Oracle Linux.

Important terms used in Ansible

Ansible Installation in Linux

Once you have compared and weighed your options and decided to go for Ansible, the next step is to have it installed on your system. We will go through the steps of installation in different Linux distributions, the most popular ones, in the next small tutorial.

Install Ansible on Centos/RedHat systems

Step 1) Install EPEL repo

[root@ansible-server ~]# sudo yum install epel-release

Step 2) Install ansible package

[root@ansible-server ~]# sudo  yum install -y ansible

Install ansible on Ubuntu/Debian systems

Step 1) Perform an update to the packages

$ sudo apt update

Step 2) Install the software-properties-common package

$ sudo apt install software-properties-common

Step 3) Install ansible personal package archive

$ sudo apt-add-repository ppa:ansible/ansible

Step 4) Install ansible

$ sudo apt update
$ sudo apt install ansible

Ansible ad-hoc commands

One of the simplest ways Ansible can be used is by using ad-hoc commands. These can be used when you want to issue some commands on a server or a bunch of servers. Ad-hoc commands are not stored for future uses but represent a fast way to interact with the desired servers.

For this Ansible tutorial, a simple two servers hosts file will be configured, containing host1 and host2.

You can make sure that the hosts are accessible from the ansible server by issuing a ping command on all hosts.

[root@ansible-server test_ansible]# ansible -i hosts all -m ping
host1 | SUCCESS => {
    "changed": false,
    "ping": "pong"
}
host2 | SUCCESS => {
    "changed": false,
    "ping": "pong"
}

Explanation:

1. Status of the command, in this case, SUCCESS
2. Host on which the command ran
3. The command issued via the -m parameter, in this case, ping
4. With the -i parameter, you can point to the hosts file.

You can issue the same command only on a specific host if needed.

[root@ansible-server test_ansible]# ansible -i hosts all -m ping --limit host2
host2 | SUCCESS => {
    "changed": false,
    "ping": "pong"
}

Explanation:

1. –Limit parameter can be used to issue commands only on specific hosts in the host's file
2. Name of the host as defined in the inventory file

If you need to copy a file to multiple destinations rapidly, you can use the copy module in ansible which uses SCP. So the command and its output look like below:

[root@ansible-server test_ansible]# ansible -i hosts all -m copy -a "src=/root/test_ansible/testfile dest=/tmp/testfile"
host1 | SUCCESS => {
    "changed": true,
    "checksum": "da39a3ee5e6b4b0d3255bfef95601890afd80709",
    "dest": "/tmp/testfile",
    "gid": 0,
    "group": "root",
    "md5sum": "d41d8cd98f00b204e9800998ecf8427e",
    "mode": "0644",
    "owner": "root",
    "size": 0,
    "src": "/root/.ansible/tmp/ansible-tmp-1562216392.43-256741011164877/source",
    "state": "file",
    "uid": 0
}
host2 | SUCCESS => {
    "changed": true,
    "checksum": "da39a3ee5e6b4b0d3255bfef95601890afd80709",
    "dest": "/tmp/testfile",
    "gid": 0,
    "group": "root",
    "md5sum": "d41d8cd98f00b204e9800998ecf8427e",
    "mode": "0644",
    "owner": "root",
    "size": 0,
    "src": "/root/.ansible/tmp/ansible-tmp-1562216392.6-280302911361278/source",
    "state": "file",
    "uid": 0
}

Explanation:

1. Copy module defined
2. Module arguments, in this case, are source absolute path and destination absolute path.
3. Ansible command output reflecting the success of the copy command and other details like the sha1 or md5 checksums for file integrity check and metadata like owner, size, or permissions.

   It is effortless to have a package installed on a bunch of servers. Ansible has several modules that interact with used installers, like yum, apt, dnf, etc.

In the next example, you will find out how to install a package via the yum module on two Centos hosts.

[root@ansible-server test_ansible]# ansible -i hosts all -m yum -a 'name=ncdu state=present'
host1 | SUCCESS => {
    "changed": true,
    "msg": "",
    "rc": 0,
    "results": [


"Loaded plugins: fastestmirror\nLoading mirror speeds from cached hostfile\n * base: mirror.netsite.dk\n * elrepo: mirrors.xservers.ro\n * epel: fedora.mirrors.telekom.ro\n * extras: centos.mirrors.telekom.ro\n * remi-php70: remi.schlundtech.de\n * remi-safe: remi.schlundtech.de\n * updates: centos.mirror.iphh.net\nResolving Dependencies\n--> Running transaction check\n---> Package ncdu.x86_64 0:1.14-1.el7 will be installed\n--> Finished Dependency Resolution\n\nDependencies Resolved\n\n================================================================================\n Package         Arch              Version                Repository       Size\n================================================================================\nInstalling:\n ncdu            x86_64            1.14-1.el7             epel             51 k\n\nTransaction Summary\n================================================================================\nInstall  1 Package\n\nTotal download size: 51 k\nInstalled size: 87 k\nDownloading packages:\nRunning transaction check\nRunning transaction test\nTransaction test succeeded\nRunning transaction\n  Installing : ncdu-1.14-1.el7.x86_64                                       1/1 \n  Verifying  : ncdu-1.14-1.el7.x86_64                                       1/1 \n\nInstalled:\n  ncdu.x86_64 0:1.14-1.el7                                                      \n\nComplete!\n"
    ]
}
host2 | SUCCESS => {
    "changed": true,
    "msg": "",
    "rc": 0,
    "results": [
        "Loaded plugins: fastestmirror\nLoading mirror speeds from cached hostfile\n * base: mirror.netsite.dk\n * elrepo: mirrors.leadhosts.com\n * epel: mirrors.nav.ro\n * extras: centos.mirrors.telekom.ro\n * remi-php70: mirrors.uni-ruse.bg\n * remi-safe: mirrors.uni-ruse.bg\n * updates: centos.mirror.iphh.net\nResolving Dependencies\n--> Running transaction check\n---> Package ncdu.x86_64 0:1.14-1.el7 will be installed\n--> Finished Dependency Resolution\n\nDependencies Resolved\n\n================================================================================\n Package         Arch              Version                Repository       Size\n================================================================================\nInstalling:\n ncdu            x86_64            1.14-1.el7             epel             51 k\n\nTransaction Summary\n================================================================================\nInstall  1 Package\n\nTotal download size: 51 k\nInstalled size: 87 k\nDownloading packages:\nRunning transaction check\nRunning transaction test\nTransaction test succeeded\nRunning transaction\n  Installing : ncdu-1.14-1.el7.x86_64                                       1/1 \n  Verifying  : ncdu-1.14-1.el7.x86_64                                       1/1 \n\nInstalled:\n  ncdu.x86_64 0:1.14-1.el7                                                      \n\nComplete!\n"
    ]
}

Explanation:

1. Yum module is used in this example
2. - It defines the module arguments, and in this case, you will choose the name of the package and its state. If the state is absent, for example, the package will be searched and if found, removed
3. When colored in yellow, you will see the output of the ansible command with the state changed, meaning in this case, that the package was found and installed.
4. Status of the yum install command issued via ansible. In this case the package ncdu.x86_64 0:1.14-1.el7 was installed.

Of course, all of the yum installer options can be used via ansible, including update, install, latest version, or remove.

In the below example the same command was issued to remove the previously installed ncdu package.

[root@ansible-server test_ansible]# ansible -i hosts all -m yum -a 'name=ncdu state=absent'
host1 | SUCCESS => {
    "changed": true,
    "msg": "",
    "rc": 0,
    "results": [
        "Loaded plugins: fastestmirror\nResolving Dependencies\n--> Running transaction check\n---> Package ncdu.x86_64 0:1.14-1.el7 will be erased\n--> Finished Dependency Resolution\n\nDependencies Resolved\n\n================================================================================\n Package         Arch              Version               Repository        Size\n================================================================================\nRemoving:\n ncdu            x86_64            1.14-1.el7            @epel             87 k\n\nTransaction Summary\n================================================================================\nRemove  1 Package\n\nInstalled size: 87 k\nDownloading packages:\nRunning transaction check\nRunning transaction test\nTransaction test succeeded\nRunning transaction\n  Erasing    : ncdu-1.14-1.el7.x86_64                                       1/1 \n  Verifying  : ncdu-1.14-1.el7.x86_64                                       1/1 \n\nRemoved:\n  ncdu.x86_64 0:1.14-1.el7                                                      \n\nComplete!\n"
    ]
}
host2 | SUCCESS => {
    "changed": true,
    "msg": "",
    "rc": 0,
    "results": [
        "Loaded plugins: fastestmirror\nResolving Dependencies\n--> Running transaction check\n---> Package ncdu.x86_64 0:1.14-1.el7 will be erased\n--> Finished Dependency Resolution\n\nDependencies Resolved\n\n================================================================================\n Package         Arch              Version               Repository        Size\n================================================================================\nRemoving:\n ncdu            x86_64            1.14-1.el7            @epel             87 k\n\nTransaction Summary\n================================================================================\nRemove  1 Package\n\nInstalled size: 87 k\nDownloading packages:\nRunning transaction check\nRunning transaction test\nTransaction test succeeded\nRunning transaction\n  Erasing    : ncdu-1.14-1.el7.x86_64                                       1/1 \n  Verifying  : ncdu-1.14-1.el7.x86_64                                       1/1 \n\nRemoved:\n  ncdu.x86_64 0:1.14-1.el7                                                      \n\nComplete!\n"
    ]
}

Explanation:

1. The output of the yum command shows that the package was removed.

Another useful and essential feature that ansible uses to interact with the client's server is to gather some facts about the system. So, it fetches hardware, software, and versioning information from the system and stores each value in a variable that can be later on used.

If you need detailed information about the systems to be modified via ansible, the next command can be used. The setup module gathers facts from the system variables.

Ansible playbook commands use YAML format, so there is not much syntax needed, but indentation must be respected. Like the name is saying, a playbook is a collection of plays. Through a playbook, you can designate specific roles to some hosts and other roles to other hosts. By doing so, you can orchestrate multiple servers in very diverse scenarios, all in one playbook.

To have all the details precise before continuing with Ansible playbook examples, we must first define a task. These are the interface to ansible modules for roles and playbooks.

Now, let's learn Ansible playbook through an example with one playbook with one play, containing multiple tasks as below:

---

- hosts: group1
  tasks:
  - name: Install lldpad package
    yum:
      name: lldpad
      state: latest
  - name: check lldpad service status
    service:
      name: lldpad
      state: started

In the above Ansible playbook example, the group1 of hosts in the host's file is targeted for lldpad package installation using the yum module and afterward the service lldpad created after the installation is then started using the service module used mostly to interact with systemd ensemble.

Explanation:

1. Group of hosts on which the playbook will run
2. Yum module is used in this task for lldpad installation
3. The service module is used to check if the service is up and running after installation

Each ansible playbook works with an inventory file. The inventory file contains a list of servers divided into groups for better control for details like IP Address and SSH port for each host.

The inventory file you can use for this Ansible playbook example looks like below. There are two groups, named group1 and group2 each containing host1 and host2 respectively.

[group1]
host1 ansible_host=192.168.100.2 ansible_ssh_port=22
[group2]
host2 ansible_host=192.168.100.3 ansible_ssh_port=22

Explanation:

1. Group name
2. Hostname, with IP address and ssh port, in this case, the default one, 22.

Another useful Ansible playbook example containing this time two plays for two hosts is the next one. For the first group of hosts, group1, selinux will be enabled. If it is enabled, then a message will appear on the screen of the host.

For the second group of hosts, httpd package will be installed only if the ansible_os_family is RedHat and ansible_system_vendor is HP.

Ansible_os_family and ansible_system_vendor are variables gathered with gather_facts option and can be used like in this conditional example.

---

- hosts: group1
  tasks:
  - name: Enable SELinux
    selinux:
      state: enabled
    when: ansible_os_family == 'Debian'
    register: enable_selinux

  - debug:
      Imsg: "Selinux Enabled. Please restart the server to apply changes."
    when: enable_selinux.changed == true

- hosts: group2
  tasks:
  - name: Install apache
    yum:
      name: httpd
      state: present
    when: ansible_system_vendor == 'HP' and ansible_os_family == 'RedHat'

Explanation:

1. Example of the when clause, In this case, when OS type is Debian. The ansible_os_family variable is gathered via gather_facts functionality.
2. The task output is registered for future use, with its name enable_selinux
3. Another example of the when clause. In this case, a message will be displayed for the host user if the SELinux was indeed enabled before.
4. Another example of the when clause consisting of two rules

Besides tasks, there are also some particular tasks called handlers. Handlers must have a unique name throughout the playbook. These work in the same way as a regular task but a handler can be notified via a notifier.

If a handler is not notified during the run of the playbook, it will not run. However, if more than one task notifies a handler, this will run only once after all the tasks are finished.

In the example shown below, you can see how a specific task has a notify section which calls upon another task. If the output of the first task is changed, then a handler task will be called. The best example is to have a configuration file changed and afterward restart that specific service.

---

- hosts: group2
  tasks:
  - name: sshd config file modify port
    lineinfile:
     path: /etc/ssh/sshd_config
     regexp: 'Port 28675'
     line: '#Port 22'
    notify:
       - restart sshd
handlers
    - name: restart sshd
      service: sshd
        name: sshd
        state: restarted
	

In this case, if the first task, "sshd config file modify port" is changed, meaning that if the port is not 28675 in the first place, then it will be modified and the task will notify the handler with the same name to run, and it will restart the sshd service.

Explanation:

1. Example of a notifier
2. Example of a handler

Ansible Roles

When dealing with extensive playbooks, it is easier to split the tasks into roles. This also helps in reusing the roles in the future. Roles are a collection of tasks, which can be moved from one playbook to another, can be run independently but only through a playbook file.

Roles are stored in separate directories and have a particular directory structure.

[root@ansible-server test2]# tree
.
`-- role1
    |-- defaults
    |   `-- main.yml
    |-- handlers
    |   `-- main.yml
    |-- meta
    |   `-- main.yml
    |-- README.md
    |-- tasks
    |   `-- main.yml
    |-- tests
    |   |-- inventory
    |   `-- test.yml
    `-- vars
        `-- main.yml

7 directories, 8 files

The yaml file in the defaults directory contains a list of default variables that are to be used along with the playbook. The handlers directory is used to store handlers. The meta-directory is supposed to have information about the author and role dependencies. In the tasks directory, there is the main yaml file for the role.

The tests directory contains a sample yaml playbook file and a sample inventory file and is mostly used for testing purposes before creating the actual role.

The vars directory contains the yaml file in which all the variables used by the role will be defined. The directory templates and the directory files should contain files and templates that will be used by the tasks in the role.

To create the directory tree for a role, you should use the following command with the last parameter, the role name:

[root@ansible-server test2]# ansible-galaxy init role1

Ansible also works well with templates. As a language for templating, it uses Jinja2.

In the next example, you will find out how a basic jinja2 template looks like and use it in a role.

At the run time, depending on, let's say in which datacenter your server is located, you can select from more than one nameservers, each corresponding to a datacenter, using the variable "resolver_ip_addresses."

{% for resolver in resolver_ip_addresses %}
nameserver {{ resolver }}
{% endfor %}

options timeout:1
options attempts:5
options rotate

In this example, in the playbook directory, there are defined some variables, including a variable named resolver_ip_addresses with different values depending on the datacenter.

- name: Set resolver for server
  template:
    src: dns.j2
    dest: /etc/resolv.conf
    group: root
    owner: root
    mode: "0644"
    tag: resolver	

Explanation:

1. Name of the template to be used. Template is located in templates dir in the role path
2. Destination path of the filename to be replaced with the template, on the client-side.
3. Permissions of the destination file

The roles tasks can also have a tag field, which has a name attributed. More than one task can share the same tag. When running an ansible playbook, you can specify the tag as well, so those tasks will be executed.

Ansible Case Study

In this section, we will analyze a Case study of an essential ansible playbook that has three roles. The purpose of this is to give a practical example of what we talked about before. Some of the examples used before in this Ansible playbook tutorial will be adapted and used in this playbook.

Below is the directory structure of the playbook. The Yaml file that will be used will be p4.yml.

[root@ansible-server test_ansible]# ls -lrth
total 16K
-rw-r--r--. 1 root root   0 Jul  3 10:13 testfile
-rw-r--r--. 1 root root 203 Jul  3 13:30 p1.yml
-rw-r--r--. 1 root root 125 Jul  3 15:00 hosts
-rw-r--r--. 1 root root 488 Jul  3 16:40 p2.yml
-rw-r--r--. 1 root root 188 Jul  4 17:33 p4.yml
drwxr-xr-x. 5 root root  47 Jul  4 17:35 roles
[root@ansible-server test_ansible]# cd roles
[root@ansible-server roles]# ls -lrth
total 12K
drwxr-xr-x. 9 root root 4.0K Jul  4 12:52 httpd
drwxr-xr-x. 9 root root 4.0K Jul  4 13:55 selinux
drwxr-xr-x. 9 root root 4.0K Jul  4 16:54 resolver

The playbook has three roles, one called resolver that sets a specific nameserver on the servers by copying a file from the server to the /etc/resolv.conf destination. Another one is called httpd, and it installs the httpd package with yum module, and the third one enables SELinux and notifies the logged user to reboot the system. Each role was created using ansible-galaxy command.

Resolver role, main.yml task:

Httpd role, main.yml task:

Selinux role, main.yml task:

Below is the p4.yml playbook defined. It will run on all hosts if not otherwise specified in the command line, it will run as root user on port 22 (SSH), it will gather facts before running the roles, and it will run all three roles mentioned before. Each role can be run independently by specifying the tag in the ansible-playbook command line with the –t parameter.

---

- hosts: all
  user: root
  port: 22
  gather_facts: True
  roles:
    - { role: selinux, tags: selinux }
    - { role: httpd, tags: httpd }
    - { role: resolver, tags: resolver }

Running the p4.yml playbook on two hosts and interpreting the output. The same command can be run with the –check parameter for a dry-run. In case you want to use password authentication, use -k parameter.

Explanation:

1. Ansible-playbook command that runs p4.yml
2. Playbook skipsSELinux role because it is already enabled.
3. Ansible found that httpd package is already installed, so it returns ok.
4. Resolver was set, and role resolver got status changed.

Ansible Commands Cheat Sheet

Install EPEL repo on Centos/RHEL systems

[root@ansible-server ~]# sudo yum install epel-release

Install ansible package on Centos/RHEL systems

[root@ansible-server ~]# sudo  yum install -y ansible

Perform an update to the packages on Debian/Ubuntu systems

$ sudo apt update

Install the software-properties-common package on Debian/Ubuntu systems

$ sudo apt install software-properties-common

Install ansible personal package archive on Debian/Ubuntu systems

$ sudo apt-add-repository ppa:ansible/ansible

Install ansible on Debian/Ubuntu systems

$ sudo apt update
$ sudo apt install ansible

Issue a ping command on all servers defined in the inventory file named hosts

 
[root@ansible-server test_ansible]# ansible -i hosts all -m ping

Issue a ping command only on host2

[root@ansible-server test_ansible]# ansible -i hosts all -m ping --limit host2

Copy the file "testfile" on all hosts in the inventory file

[root@ansible-server test_ansible]# ansible -i hosts all -m copy -a "src=/root/test_ansible/testfile dest=/tmp/testfile"

Install ncdu package on all hosts

[root@ansible-server test_ansible]# ansible -i hosts all -m yum -a 'name=ncdu state=present'

Remove ncdu package on all hosts

[root@ansible-server test_ansible]# ansible -i hosts all -m yum -a 'name=ncdu state=absent'

Build the directory structure for role named role1.

[root@ansible-server test2]# ansible-galaxy init role1

Dry-run p4.yml playbook

[root@ansible-server test_ansible]# ansible-playbook -i hosts p4.yml --check

Run p4.yml playbook with password authentication for all hosts

[root@ansible-server test_ansible]# ansible-playbook -i hosts p4.yml -k

Summary

In a world with technology that is continuously changing at a swift pace and growing incredibly fast at the same time, system administrators and devops engineers must think of different approaches on how to automate routine tasks and orchestrate large pools of servers.

While there are many tools (Chef, Puppet) out there that do the same thing with some differences, Ansible managed to rise above all of them with its simplicity, improved security, and most important its smooth learning curve. Due to these qualities and fast adoption of Ansible, we created a tutorial full of examples so you can have an even more seamless first experience in working with Ansible.

In this Ansible basics tutorial, we described ansible and talked a bit about its history. We mentioned the strong points of Ansible and the advantages that ansible can bring to automation and orchestration of infrastructures of different sizes. We defined the essential ansible used terms and defined the structure of Ansible playbooks. Thorough examples accompanied all information with detailed explanations.

Patrick Cockwell

Jul 24, 2019 · 7 min read

At FLYR we use Apache Airflow for many of our scheduled data ingestion jobs. Airflow is a scheduler and monitor with a graphical interface that displays workflows as Directed Acyclic Graphs (DAGs).

We’ve learned through trial and error that Airflow is great in some situations, but has concrete limitations that make it inefficient to use for certain applications. In this article we’ll discuss the benefits of using Airflow, solutions to common issues, and situations in which Airflow might not be the best tool to use for implementation.

Airflow is likely one of the best open-source schedulers available for straightforward ETL tasks. It has a gentle learning curve for simplistic tasks because it uses Python and is fast to start up, making it a great tool when you need to quickly set up a scheduler.

Airflow is an open-source tool, meaning that its source code is available to anyone to use or modify. In other words, Airflow can be branched from its GitHub repository and adjusted to fit a company’s unique workflow. In addition, Airflow has an open-source community that submits improvements to the code base. Companies using Airflow can also potentially submit code improvements, and one of FLYR’s own employees has contributed to the Airflow project.

Airflow’s DAGs are relatively simple to create and efficient to use once they’ve been set up. After a DAG has been created, Airflow’s GUI shows a visual history of workloads and readable logs, which makes debugging easier. The DAGs are also defined in Python, making the learning curve to use them gentle for many startup engineers.

DAGs in Airflow can be scheduled using multiple methods, whether that is for regular time intervals without a concrete starting point, according to CRON schedules, or just through one-time execution by using an @once tag. DAGs also support non-time based triggering (by simply omitting a schedule), in which case the DAG has to be triggered manually or by an external actor, whether that’s through another DAG, an HTTP request from a service, or by some other method.

Though backfilling is primarily relevant to ETL processes, it can be useful for other reasons as well. Backfilling allows Airflow to replay tasks as though the task were being run in the past. Backfilled tasks can be set to be dependent upon previous executions of DAG tasks or can be set to run when resources become available.

When using backfilling, it’s best to make sure your workload ensures data idempotency. Data idempotency is the guarantee that running a job will have the same end result regardless of the initial state of the data’s destination. In other words, as long as a job is data idempotent, the time at which you run the job won’t affect the results your job will produce.

While Airflow is easy to set up when your use cases are simple, implementing complex tasks require some in-depth knowledge of how Airflow works. In this section, we’ll talk about common situations you might come across and some ways to efficiently implement Airflow.

Because Airflow is open-source, you can extend the source code when you need Airflow to handle data formats it doesn’t inherently support. To do this, you will likely want to create subclass operators. Extending Airflow with subclass operators will allow you to avoid needing to copy boilerplate code in order to support your data format. Subclassing your operators will also improve the rigor and consistency of your nonstandard data formats, though it requires a bit more work.

Though Airflow has a quick learning curve, it also has some features that may make developers do a double take. Instead of logging the DAG process execution time when the DAG was started, Airflow logs the time as the start time minus one execution cycle, which is the amount of time in between DAG triggers. This means that DAG execution times are often recorded to have occurred in the past rather than at the actual time they began. While this isn’t detrimental to Airflow’s functionality as a scheduler, it can cause confusion for developers unfamiliar with the interface.

Sensors take up a worker slot during their whole run by default, which means they use process threads until the task they are checking for completes. An unfortunate consequence of this default design is that the sensors and tasks can end up in a deadlock. The task that the sensor is checking on might be waiting for threads to be freed; however, the sensor task may be taking up the very thread resources the task it is waiting on requires to run. This can be resolved by setting the sensor to “reschedule” mode, which spins down the sensor completely and reschedules it to check for the task state at a later time.

Though DAGs function relatively well when they are running isolated tasks, as soon as a DAG’s trigger is dependent on another DAG, unexpected problems can occur. We’ll discuss how cross-DAG communication works, some common issues you might encounter, and solutions to those problems.

Airflow has a built-in cross-DAG communication tool called xcom that is similar to a Redis store in that it stores information as key and value pairs. The key values are often ID values that originate directly from a DAG and are stored in a PostGres database. This design works well when tasks are implemented using normal DAGs; however, a major detriment of xcom is that it doesn’t allow for DAG-to-subdag communication and vice versa. To better understand why this is the case, we’ll need to delve into the technicalities of how subdags function.

A DAG can be divided into tasks. It is possible for one of these tasks to be a DAG in and of itself — that type of task is called a subdag. Subdags have a limitation in the basic implementation of Airflow in that xcom cannot receive or pass messages to a parent DAG. Since subdags cannot access values from a normal DAG, they also can’t access xcom values that are stored using DAG IDs.

There are at least two potential solutions to this issue: Airflow can be extended to allow subdags to communicate with normal DAGs, or an external storage can be used to store artifacts.

DAG-to-DAG trigger communication through xcom also provides no traceability, so a more reliable method of triggering DAGs through other DAGs is by using task sensors. These task sensors will sit and wait for the completion of a DAG at a specific UTC date time. Though these sensors function better than xcom in some ways, they too have some caveats to consider. Sensors will only check for completion of a task at a specific UTC timestamp⁠ — they cannot look for completion within a timespan.

Airflow DAGs have two phases to their lifecycle: initialization and execution. Though execution only occurs when the DAG is triggered, reinitialization occurs every thirty seconds, which results in shared, generated variables being overwritten. To address this, you can store shared data in external sources as long as the data is not dynamic.

For example, any attempts to timestamp when a UUID was generated by a DAG will fail because every time the reinitialization phase is run, the DAG will generate a new UUID. In addition, any DAGs that should share variables will become mismatched over time as the DAG reinitializes. Instead of storing these within the DAGs, the data/artifacts should be stored outside of Airflow.

You should not, however, use this method to store dynamic data since it will result in mismatches between the stored data and the data expected by the DAGs. For example, if the number of tasks run by a DAG is stored in an external database and that number changes while the tasks are running, Airflow will reinitialize the DAG according to the number stored in the database rather than the actual number of running tasks. This can result in unknown behavior during execution.

When you have a hammer, everything looks like a nail. Though Airflow is a useful tool for ETL processes, it’s easy to fall prey to the same mindset and to believe Airflow is the solution for everything.

While using Airflow, an important fact to keep in mind is that Airflow is a scheduler and is not designed for distributed computation. Airflow requires at least one worker to be up and running at any point in time, regardless of whether the worker is actually being used and these inefficient worker costs can add up when you use Airflow for many tasks. We explored this problem in more detail in another blog post that can be read here.

Airflow is an excellent scheduler to use for ETL tasks. It is free and one of the quickest ways to immediately implement a scheduler.

When using Airflow for complex tasks, make sure to put significant forethought into the design of any DAGs to ensure the tasks run as anticipated. Data should be set up so that it is idempotent in order to guarantee jobs will output the correct data and backfilling will work as expected. When Airflow is used in a data engineering capacity, we recommend following a Git paradigm for managing code. For automated CI/CD pipelines, normal code deployment paradigms are likely sufficient.

Our data engineering team works extensively with Airflow to implement ETL pipelines for airline data ingestion. We’re currently hiring.

This section describes all the steps to build the DAG shown in figure 1.

As you know, spark-submit script is used for submitting an Spark app to an Spark cluster manager. For example, running PySpark app search_event_ingestor.py is as follows:

spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.0.0,io.delta:delta-core_2.12:0.7.0 --master local[*] --driver-memory 12g --executor-memory 12g spark/search_event_ingestor.py

So for building an SparkSubmitOperator in Airflow you need to do the followings:

3–1. SPARK_HOME environment variable — We need to set spark binary dir in OS environment variable as follows (in Ubuntu):
export SPARK_HOME=/path_to_the_spark_home_dir
export PATH=$PATH:$SPARK_HOME/bin

3–2. Spark Connection — Create Spark connection in Airflow web ui (localhost:8080) > admin menu > connections > add+ > Choose Spark as the connection type, give a connection id and put the Spark master url (i.e local[*], or the cluster manager master’s URL) and also port of your Spark master or cluster manager if you have an Spark cluster.

3–3. Set Spark app home variable — This is very useful to define a global variable in Airflow to be used in any DAG. I define the PySpark app home dir as an Airflow variable which will be used later. In admin menu, hit the variable and define the variable as shown in the figure below:

3–4. Building DAG — Now, it’s time to build an Airflow DAG. As I said earlier, an Airflow DAG is a typical Python script which needs to be in the dags_folder(This is a configuration option in airflow.cfg).

Let’s get into it our DAG file named flight_search_dag.py:

3–4–1 Imports — Like any other Python app, we need to import some modules and classes as follows:

from datetime import datetime, timedelta
import pendulum
from airflow import DAG
from airflow.contrib.operators.spark_submit_operator import SparkSubmitOperator
from airflow.models import Variable

3–4–2 default_args — This will be a dictionary for setting up default configuration of our DAG. (This is almost the same for all DAGs):

local_tz = pendulum.timezone("Asia/Tehran")
default_args = {
'owner': 'mahdyne',
'depends_on_past': False,
'start_date': datetime(2020, 10, 10, tzinfo=local_tz),
'email': ['nematpour.ma@gmail.com'],
'email_on_failure': True,
'email_on_retry': True,
'retries': 2,
'retry_delay': timedelta(minutes=5)
}

3–4–3 dag — This is going to be an object instantiated using DAG class:

dag = DAG(dag_id='flight_search_dag',
default_args=default_args,
catchup=False,
schedule_interval="0 * * * *")

catchup=False means we do not need Airflow to fill the undone past execution since the start_date. schedule_interval="0 * * * *" You guessed right! We can pass contab-style scheduling pattern to this attribute.

3–4–4 pyspark_app_home — This variable is set for keeping the PySpark app dir as defined in Airflow Variable in the UI earlier:

pyspark_app_home=Variable.get("PYSPARK_APP_HOME")

3–4–5 SparkSubmitOperator — We can have a Spark app in a DAG(i.e. a task which executes Spark app in theDAG) using This kind of operator very simple and straightforward.

flight_search_ingestion= SparkSubmitOperator(task_id='flight_search_ingestion',
conn_id='spark_local',
application=f'{pyspark_app_home}/spark/search_event_ingestor.py',
total_executor_cores=4,
packages="io.delta:delta-core_2.12:0.7.0,org.apache.spark:spark-sql-kafka-0-10_2.12:3.0.0",
executor_cores=2,
executor_memory='5g',
driver_memory='5g',
name='flight_search_ingestion',
execution_timeout=timedelta(minutes=10),
dag=dag
)

The SparkSubmitOperator class includes useful attributes that eliminate the need for a separate bash script and calling it using a BashOperator.

conn_id attribute takes the name of Spark connection which has been built in section 3.2. We can pass the path of the PySpark app Python file to the application attribute, and also pass the dependencies using packages in comma separated style. Other attributes are Spark developer-friendly :).

This is easy as well in the Spark Scala/Java case. You would pass the fat jar file to the application attribute and also the pass the main class to the attribute jar_class.

There are two other SparkSubmitOperator tasks like flight_search_ingestion named flight_search_waiting_time, flight_nb_search.

3–4–6 Dependecies — After instantiating other tasks, now is the time to define the dependencies.

flight_search_ingestion>>[flight_search_waiting_time,flight_nb_search]

Airflow is overloading the binary right shift >> oparator to define the dependencies, meaning that flight_search_ingestion should be executed successfully first and then two tasks flight_search_waiting_time, flight_nb_search are run in parallel since these two tasks both depend on the first task flight_search_ingestion but do not depend on each other and also we have enough resources in the cluster to run two Spark jobs at the same time.

Finished! Now we have a DAG including three Spark jobs that is running once an hour and receive email if something goes wrong(i.e. any failure or running longer than expected). Don’t forget to turn on the DAG by the cool button above :) .

The project can be found on my Github repository https://github.com/mahdyne/pyspark-tut

Getting Started

Embedded Cassandra provides an easy way to start and stop Apache Cassandra.

To learn more about Embedded Cassandra, please consult the reference documentation.

All versions of Embedded Cassandra reference documentation are here.

Here is a quick teaser of starting Cassandra:

import java.net.InetSocketAddress;
import com.datastax.oss.driver.api.core.CqlSession;
import com.github.nosan.embedded.cassandra.Cassandra;
import com.github.nosan.embedded.cassandra.CassandraBuilder;
import com.github.nosan.embedded.cassandra.Settings;
import com.github.nosan.embedded.cassandra.cql.CqlScript;
public class CassandraExamples {
	public static void main(String[] args) {
		Cassandra cassandra = new CassandraBuilder().build();
		cassandra.start();
		try {
			Settings settings = cassandra.getSettings();
			try (CqlSession session = CqlSession.builder()
					.addContactPoint(new InetSocketAddress(settings.getAddress(), settings.getPort()))
					.withLocalDatacenter("datacenter1")
					.build()) {
				CqlScript.ofClassPath("schema.cql").forEachStatement(session::execute);
			}
		}
		finally {
			cassandra.stop();
		}
	}
}

Issues

Embedded Cassandra uses GitHub's issue tracking system to report bugs and feature requests. If you want to raise an issue, please follow this link

Also see CONTRIBUTING.md if you wish to submit pull requests.

Build

Embedded Cassandra can be easily built with the maven wrapper. You also need JDK 1.8.

License

Embedded Cassandra is released under the Apache License 2.0

Extensions

Simple I/O for Parquet. Allows you to easily read and write Parquet files in Scala.

Use just a Scala case class to define the schema of your data. No need to use Avro, Protobuf, Thrift or other data serialisation systems. You can use generic records if you don't want to use the case class, too.

Compatible with files generated with Apache Spark. However, unlike in Spark, you do not have to start a cluster to perform I/O operations.

Based on official Parquet library, Hadoop Client and Shapeless.

Integrations for Akka Streams and FS2.

Released for Scala 2.11.x, 2.12.x and 2.13.x. FS2 integration is available for 2.12.x and 2.13.x.

Tutorial

1. Quick Start
2. AWS S3
3. Akka Streams
4. FS2
5. Before-read filtering or filter pushdown
6. Schema projection
7. Statistics
8. Supported storage types
9. Supported types
10. Generic Records
11. Customisation and Extensibility
12. More Examples
13. Contributing

Quick Start

SBT

libraryDependencies ++= Seq(
  "com.github.mjakubowski84" %% "parquet4s-core" % "1.8.1",
  "org.apache.hadoop" % "hadoop-client" % yourHadoopVersion
)

Mill

def ivyDeps = Agg(
  ivy"com.github.mjakubowski84::parquet4s-core:1.8.1",
  ivy"org.apache.hadoop:hadoop-client:$yourHadoopVersion"
)

import com.github.mjakubowski84.parquet4s.{ ParquetReader, ParquetWriter }
case class User(userId: String, name: String, created: java.sql.Timestamp)
val users: Iterable[User] = Seq(
  User("1", "parquet", new java.sql.Timestamp(1L))
)
val path = "path/to/local/parquet"
// writing
ParquetWriter.writeAndClose(path, users)
// reading
val parquetIterable = ParquetReader.read[User](path)
try {
  parquetIterable.foreach(println)
} finally parquetIterable.close()

AWS S3

In order to connect to AWS S3 you need to define one more dependency:

"org.apache.hadoop" % "hadoop-aws" % yourHadoopVersion

Next, the most common way is to define following environmental variables:

export AWS_ACCESS_KEY_ID=my.aws.key
export AWS_SECRET_ACCESS_KEY=my.secret.key

Please follow documentation of Hadoop AWS for more details and troubleshooting.

Passing Hadoop Configs Programmatically

File system configs for S3, GCS or Hadoop can also be set programmatically to the ParquetReader and ParquetWriter by passing the Configuration object to the ParqetReader.Options and ParquetWriter.Options case classes.

Akka Streams

Parquet4S has an integration module that allows you to read and write Parquet files using Akka Streams. Just import:

"com.github.mjakubowski84" %% "parquet4s-akka" % "1.8.1"
"org.apache.hadoop" % "hadoop-client" % yourHadoopVersion

Parquet4S has so far a single Source for reading single file or directory and Sinks for writing.

import com.github.mjakubowski84.parquet4s.{ParquetStreams, ParquetWriter}
import org.apache.parquet.hadoop.ParquetFileWriter
import org.apache.parquet.hadoop.metadata.CompressionCodecName
import akka.actor.ActorSystem
import akka.stream.{ActorMaterializer, Materializer}
import akka.stream.scaladsl.Source
import org.apache.hadoop.conf.Configuration
import scala.concurrent.duration._
case class User(userId: String, name: String, created: java.sql.Timestamp)
implicit val system: ActorSystem = ActorSystem()
implicit val materializer: Materializer = ActorMaterializer()
val users: Iterable[User] = ???
val conf: Configuration = ??? // Set Hadoop configuration programmatically
// Please check all the available configuration options!
val writeOptions = ParquetWriter.Options(
  writeMode = ParquetFileWriter.Mode.OVERWRITE,
  compressionCodecName = CompressionCodecName.SNAPPY,
  hadoopConf = conf // optional hadoopConf
)
// Writes a single file.
Source(users).runWith(ParquetStreams.toParquetSingleFile(
  path = "file:///data/users/user-303.parquet",
  options = writeOptions
))
// Tailored for writing indefinite streams.
// Writes file when chunk reaches size limit or defined time period elapses.
// Can also partition files!
// Check all the parameters and example usage in project sources.
Source(users).via(
  ParquetStreams
    .viaParquet[User]("file:///data/users")
    .withMaxCount(writeOptions.rowGroupSize)
    .withMaxDuration(30.seconds)
    .withWriteOptions(writeOptions)
    .build()
).runForeach(user => println(s"Just wrote user ${user.userId}..."))
  
// Reads a file or files from the path. Please also have a look at the rest of parameters.
ParquetStreams.fromParquet[User]
  .withOptions(ParquetReader.Options(hadoopConf = conf))
  .read("file:///data/users")
  .runForeach(println)

FS2

FS2 integration allows you to read and write Parquet using functional streams. Functionality is exactly the same as in case of Akka module. In order to use it please import:

"com.github.mjakubowski84" %% "parquet4s-fs2" % "1.8.1"
"org.apache.hadoop" % "hadoop-client" % yourHadoopVersion

Please check examples to learn more.

Before-read filtering or filter pushdown

One of the best features of Parquet is an efficient way of fitering. Parquet files contain additional metadata that can be leveraged to drop chunks of data without scanning them. Parquet4S allows do define filter predicates in all modules in order to push filtering out from Scala collections and Akka or FS2 stream down to point before file content is even read.

You define you filters using simple algebra as follows.

In core library:

ParquetReader.read[User](path = "file://my/path", filter = Col("email") === "user@email.com")

In Akka filter applies both to content of files and partitions:

ParquetStreams.fromParquet[Stats]
  .withFilter(Col("stats.score") > 0.9 && Col("stats.score") <= 1.0)
  .read("file://my/path")

You can construct filter predicates using ===, !==, >, >=, <, <=, and in operators on columns containing primitive values. You can combine and modify predicates using &&, || and ! operators. in looks for values in a list of keys, similar to SQL's in operator. Mind that operations on java.sql.Timestamp and java.time.LocalDateTime are not supported as Parquet still not allows filtering by Int96 out of the box.

Check ScalaDoc and code for more!

Schema projection

Schema projection is another way of optimization of reads. By default Parquet4S reads the whole content of each Parquet record even when you provide a case class that maps only a part of the columns. Such a behaviour is expected because you may want to use generic records to process your data. However, you can explicitely tell Parquet4S to use the provided case class (or implicit ParquetSchemaResolver) as an override for the original file schema. In effect, all columns not matching your schema will be skipped and not read. This functionality is available in every module of Parquet4S.

// core
ParquetReader.withProjection[User].read(path = "file://my/path")
// akka
ParquetStreams.fromParquet[User].withProjection.read("file://my/path")
// fs2
import com.github.mjakubowski84.parquet4s.parquet._
fromParquet[IO, User].projection.read(blocker, "file://my/path")

Statistics

Parquet4S leverages Parquet metadata to efficiently read record count as well as max and min value of the column of Parquet files. It provides correct value for both filtered and unfiltered files. Functionality is available in core module either by direct call to Stats or via API of ParquetReader and ParquetIterable.

Supported storage types

As it is based on Hadoop Client, Parquet4S can read and write from a variety of file systems:

• Local files
• HDFS
• Amazon S3
• Google Storage
• Azure
• OpenStack

Please refer to Hadoop Client documentation or your storage provider to check how to connect to your storage.

Supported types

Primitive types

Complex Types

Complex types can be arbitrarily nested.

• Option
• List
• Seq
• Vector
• Set
• Array - Array of bytes is treated as primitive binary
• Map - Key must be of primitive type, only immutable version.
• Since 1.2.0. Any Scala collection that has Scala 2.13 collection Factory (in 2.11 and 2.12 it is derived from CanBuildFrom). Refers to both mutable and immutable collections. Collection must be bounded only by one type of element - because of that Map is supported only in immutable version (for now).
• Any case class

Generic Records

You may want to not use strict schema and process your data in a generic way. Since version 1.2.0 Parquet4S has rich API that allows to build, transform, write and read Parquet records in easy way. Each implementation of ParquetRecord is Scala Iterable and a mutable collection. You can execute operations on RowParquetRecord and ListParquetRecord as on mutable Seq and you can treat MapParquetRecord as mutable Map. Moreover, records received addition functions like get and add (and more) that take implicit ValueCodec and allow to read and modify records using regular Scala types. There is default ParquetRecordEndcoder, ParquetRecordDecoder and ParquetSchemaResolver for RowParquetRecord so reading Parquet in a generic way works out of the box! In order to write you still need to provide a schema in form of Parquet's MessageType.

Funcionality is available in all modules. See examples.

Customisation and Extensibility

Parquet4S is built using Scala's type class system. That allows you to extend Parquet4S by defining your own implementations of its type classes.

For example, you may define your codecs of your own type so that they can be read from or written to Parquet. Assume that you have your own type:

case class CustomType(i: Int)

You want to save it as optional Int. In order to achieve you have to define your own codec:

import com.github.mjakubowski84.parquet4s.{OptionalValueCodec, IntValue, Value}
implicit val customTypeCodec: OptionalValueCodec[CustomType] = 
  new OptionalValueCodec[CustomType] {
    override protected def decodeNonNull(value: Value, configuration: ValueCodecConfiguration): CustomType = value match {
      case IntValue(i) => CustomType(i)
    }
    override protected def encodeNonNull(data: CustomType, configuration: ValueCodecConfiguration): Value =
      IntValue(data.i)
}

Additionally, if you want to write your custom type, you have to define the schema for it:

import org.apache.parquet.schema.{OriginalType, PrimitiveType}
import com.github.mjakubowski84.parquet4s.ParquetSchemaResolver._
implicit val customTypeSchema: TypedSchemaDef[CustomType] =
  typedSchemaDef[CustomType](
    PrimitiveSchemaDef(
      primitiveType = PrimitiveType.PrimitiveTypeName.INT32,
      required = false,
      originalType = Some(OriginalType.INT_32)
    )
  )

More Examples

Please check examples where you can find simple code covering basics for core, akka and fs2 modules.

Moreover, examples contain two simple applications comprising Akka Streams or FS2 and Kafka. It shows how you can write partitioned Parquet files with data coming from an indefinite stream.

Contributing

Do you want to contribute? Please read the contribution guidelines.

A curated list of Docker Compose samples.

These samples provide a starting point for how to integrate different services using a Compose file and to manage their deployment with Docker Compose.

Note:

The following samples are intended for use in local development environments such as project setups, tinkering with software stacks, etc. These samples must not be deployed in production environments.

Contents

• Samples of Docker Compose applications with multiple integrated services.
• Single service samples.
• Basic setups for different platforms (not production ready - useful for personal use).

Samples of Docker Compose applications with multiple integrated services

• ASP.NET / MS-SQL - Sample ASP.NET core application with MS SQL server database.
• Elasticsearch / Logstash / Kibana - Sample Elasticsearch, Logstash, and Kibana stack.
• Go / NGINX / MySQL - Sample Go application with an Nginx proxy and a MySQL database.
• Go / NGINX / PostgreSQL - Sample Go application with an Nginx proxy and a PostgreSQL database.
• Java Spark / MySQL - Sample Java application and a MySQL database.
• NGINX / Flask / MongoDB - Sample Python/Flask application with Nginx proxy and a Mongo database.
• NGINX / Flask / MySQL - Sample Python/Flask application with an Nginx proxy and a MySQL database.
• NGINX / Go - Sample Nginx proxy with a Go backend.
• React / Spring / MySQL - Sample React application with a Spring backend and a MySQL database.
• React / Express / MySQL - Sample React application with a Node.js backend and a MySQL database.
• React / Express / MongoDB - Sample React application with a Node.js backend and a Mongo database.
• React / Rust / PostgreSQL - Sample React application with a Rust backend and a Postgres database.
• Spring / PostgreSQL - Sample Java application with Spring framework and a Postgres database.

Single service samples

Basic setups for different platforms (not production ready - useful for personal use)

Getting started

These instructions will get you through the bootstrap phase of creating and deploying samples of containerized applications with Docker Compose.

Prerequisites

• Make sure that you have Docker and Docker Compose installed
  • Windows or macOS: Install Docker Desktop
  • Linux: Install Docker and then Docker Compose
• Download some or all of the samples from this repository.

Running a sample

The root directory of each sample contains the docker-compose.yaml which describes the configuration of service components. All samples can be run in a local environment by going into the root directory of each one and executing:

docker-compose up -d

Check the README.md of each sample to get more details on the structure and what is the expected output. To stop and remove all containers of the sample application run:

docker-compose down

Contribute

We welcome examples that help people understand how to use Docker Compose for common applications. Check the Contribution Guide for more details.

Laravel Vapor is a serverless deployment platform for Laravel, powered by AWS. Launch your Laravel infrastructure on Vapor and fall in love with the scalable simplicity of serverless.

Sign Up Now Documentation

A demo application to illustrate how Inertia.js works, ported to Symfony from Laravel.

Tested on both PHP 7.4 and 8.0.

Installation

Make sure you have the symfony binary (Symfony CLI) installed and in your PATH.

Clone the repo locally:

git clone https://github.com/aleksblendwerk/pingcrm-symfony.git pingcrm-symfony
cd pingcrm-symfony

Install dependencies:

composer install
yarn install

Build assets:

yarn build

The current configuration uses MySQL. Adjust the DATABASE_URL in .env accordingly (or optionally create a .env.local file and put your overrides there).

Create the database, schema and load the initial data:

composer build-database

Run the dev server:

symfony serve

You're ready to go! Visit Ping CRM in your browser, and login with:

• Username: johndoe@example.com
• Password: secret

Running tests

Keep in mind to adjust the DATABASE_URL in .env.test accordingly (or optionally create a .env.test.local file and put your overrides there).

Run the Ping CRM tests:

composer test

Remarks

One of the goals for this port was to leave the original JS side of things unchanged. This promise has been kept, aside from one or two very minor changes. As a result, the PHP backend code occasionally has to jump through a few hoops to mimic the expected response data formats which are partly catered to Laravel's out-of-the-box features.

Also, I am currently not really satisfied with the whole validation workflow, this might eventually get an overhaul.

Consider this a proof of concept, I am sure there is room for improvements. If any fellow Symfony developers want to join in to tackle things in more concise or elegant ways, let's go for it!

Credits

Shout-outs to all Ping CRMs all over the world!

Inertia is a new approach to building classic server-driven web apps. We call it the modern monolith.

Inertia allows you to create fully client-side rendered, single-page apps, without much of the complexity that comes with modern SPAs. It does this by leveraging existing server-side frameworks.

Inertia has no client-side routing, nor does it require an API. Simply build controllers and page views like you've always done!

See the who is it for and how it works pages to learn more.

Not a framework

Inertia isn't a framework, nor is it a replacement to your existing server-side or client-side frameworks. Rather, it's designed to work with them. Think of Inertia as glue that connects the two. Inertia does this via adapters. We currently have three official client-side adapters (React, Vue, and Svelte) and two server-side adapters (Laravel and Rails).

Join the newsletter

Building modern web apps is hard. Tools like Vue and React are extremely powerful, but the complexity they add to a full-stack developer's workflow is insane.
It doesn’t have to be this way...

Ok, I'm listening...

Say hello to Livewire.

Hi Livewire!

Livewire is a full-stack framework for Laravel that makes building dynamic interfaces simple, without leaving the comfort of Laravel.

Consider my interest piqued

It's not like anything you've seen before. The best way to understand it is to just look at the code. Strap on your snorkel, we're diving in.

...I'll get my floaties

Our Lovely Sponsors:

Pulumi's Infrastructure as Code SDK is the easiest way to create and deploy cloud software that use containers, serverless functions, hosted services, and infrastructure, on any cloud.

Simply write code in your favorite language and Pulumi automatically provisions and manages your AWS, Azure, Google Cloud Platform, and/or Kubernetes resources, using an infrastructure-as-code approach. Skip the YAML, and use standard language features like loops, functions, classes, and package management that you already know and love.

For example, create three web servers:

let aws = require("@pulumi/aws");
let sg = new aws.ec2.SecurityGroup("web-sg", {
    ingress: [{ protocol: "tcp", fromPort: 80, toPort: 80, cidrBlocks: ["0.0.0.0/0"]}],
});
for (let i = 0; i < 3; i++) {
    new aws.ec2.Instance(`web-${i}`, {
        ami: "ami-7172b611",
        instanceType: "t2.micro",
        securityGroups: [ sg.name ],
        userData: `#!/bin/bash
            echo "Hello, World!" > index.html
            nohup python -m SimpleHTTPServer 80 &`,
    });
}

Or a simple serverless timer that archives Hacker News every day at 8:30AM:

const aws = require("@pulumi/aws");
const snapshots = new aws.dynamodb.Table("snapshots", {
    attributes: [{ name: "id", type: "S", }],
    hashKey: "id", billingMode: "PAY_PER_REQUEST",
});
aws.cloudwatch.onSchedule("daily-yc-snapshot", "cron(30 8 * * ? *)", () => {
    require("https").get("https://news.ycombinator.com", res => {
        let content = "";
        res.setEncoding("utf8");
        res.on("data", chunk => content += chunk);
        res.on("end", () => new aws.sdk.DynamoDB.DocumentClient().put({
            TableName: snapshots.name.get(),
            Item: { date: Date.now(), content },
        }).promise());
    }).end();
});

Many examples are available spanning containers, serverless, and infrastructure in pulumi/examples.

Pulumi is open source under the Apache 2.0 license, supports many languages and clouds, and is easy to extend. This repo contains the pulumi CLI, language SDKs, and core Pulumi engine, and individual libraries are in their own repos.

Welcome

• Getting Started: get up and running quickly.

• Tutorials: walk through end-to-end workflows for creating containers, serverless functions, and other cloud services and infrastructure.

• Examples: browse a number of useful examples across many languages, clouds, and scenarios including containers, serverless, and infrastructure.

• Reference Docs: read conceptual documentation, in addition to details on how to configure Pulumi to deploy into your AWS, Azure, or Google Cloud accounts, and/or Kubernetes cluster.

• Community Slack: join us over at our community Slack channel. Any and all discussion or questions are welcome.

• Roadmap: check out what's on the roadmap for the Pulumi project over the coming months.

Getting Started

See the Get Started guide to quickly get started with Pulumi on your platform and cloud of choice.

Otherwise, the following steps demonstrate how to deploy your first Pulumi program, using AWS Serverless Lambdas, in minutes:

1. Install:

   To install the latest Pulumi release, run the following (see full installation instructions for additional installation options):

   $ curl -fsSL https://get.pulumi.com/ | sh

2. Create a Project:

   After installing, you can get started with the pulumi new command:

   $ mkdir pulumi-demo && cd pulumi-demo
   $ pulumi new hello-aws-javascript

   The new command offers templates for all languages and clouds. Run it without an argument and it'll prompt you with available projects. This command created an AWS Serverless Lambda project written in JavaScript.

3. Deploy to the Cloud:

   Run pulumi up to get your code to the cloud:

   $ pulumi up

   This makes all cloud resources needed to run your code. Simply make edits to your project, and subsequent pulumi ups will compute the minimal diff to deploy your changes.

4. Use Your Program:

   Now that your code is deployed, you can interact with it. In the above example, we can curl the endpoint:

   $ curl $(pulumi stack output url)

5. Access the Logs:

   If you're using containers or functions, Pulumi's unified logging command will show all of your logs:

   $ pulumi logs -f

6. Destroy your Resources:

   After you're done, you can remove all resources created by your program:

   $ pulumi destroy -y

To learn more, head over to pulumi.com for much more information, including tutorials, examples, and details of the core Pulumi CLI and programming model concepts.

Platform

CLI

Architecture	Build Status
Linux/macOS x64
Windows x64

Languages

Language	Status	Runtime
	JavaScript	Stable	Node.js 10+
	TypeScript	Stable	Node.js 10+
	Python	Stable	Python 3.6+
	Go	Stable	Go 1.13.x
	.NET (C#/F#/VB.NET)	Stable	.NET Core 3.1

Clouds

See Supported Clouds for the full list of supported cloud and infrastructure providers.

Contributing

Please See CONTRIBUTING.md for information on building Pulumi from source or contributing improvements.

PulumiUP: Hear from technical leaders as they present the vision for the future of cloud engineering. Save Your Spot

JSON to JSON transformation library written in Java where the "specification" for the transform is itself a JSON document.

Useful For

1. Transforming JSON data from ElasticSearch, MongoDb, Cassandra, etc before sending it off to the world
2. Extracting data from a large JSON documents for your own consumption

Table of Contents

1. Overview
2. Documentation
3. Shiftr Transform DSL
4. Demo
5. Getting Started
6. Getting Transform Help
7. Why Jolt Exists
8. Alternatives
9. Performance
10. CLI
11. Code Coverage
12. Release Notes

Overview

Jolt :

• provides a set of transforms, that can be "chained" together to form the overall JSON to JSON transform.
• focuses on transforming the structure of your JSON data, not manipulating specific values
  • The idea being: use Jolt to get most of the structure right, then write code to fix values
• consumes and produces "hydrated" JSON : in-memory tree of Maps, Lists, Strings, etc.
  • use Jackson (or whatever) to serialize and deserialize the JSON text

Stock Transforms

The Stock transforms are:

shift       : copy data from the input tree and put it the output tree
default     : apply default values to the tree
remove      : remove data from the tree
sort        : sort the Map key values alphabetically ( for debugging and human readability )
cardinality : "fix" the cardinality of input data.  Eg, the "urls" element is usually a List, but if there is only one, then it is a String

Each transform has its own DSL (Domain Specific Language) in order to facilitate its narrow job.

Currently, all the Stock transforms just effect the "structure" of the data. To do data manipulation, you will need to write Java code. If you write your Java "data manipulation" code to implement the Transform interface, then you can insert your code in the transform chain.

The out-of-the-box Jolt transforms should be able to do most of your structural transformation, with custom Java Transforms implementing your data manipulation.

Documentation

Jolt Slide Deck : covers motivation, development, and transforms.

Javadoc explaining each transform DSL :

• shift
• default
• remove
• cardinality
• sort
• full qualified Java ClassName : Class implements the Transform or ContextualTransform interfaces, and can optionally be SpecDriven (marker interface)
  • Transform interface
  • SpecDriven
    • where the "input" is "hydrated" Java version of your JSON Data

Running a Jolt transform means creating an instance of Chainr with a list of transforms.

The JSON spec for Chainr looks like : unit test.

The Java side looks like :

Chainr chainr = JsonUtils.classpathToList( "/path/to/chainr/spec.json" );
Object input = elasticSearchHit.getSource(); // ElasticSearch already returns hydrated JSon
Object output = chainr.transform( input );
return output;

Shiftr Transform DSL

The Shiftr transform generally does most of the "heavy lifting" in the transform chain. To see the Shiftr DSL in action, please look at our unit tests (shiftr tests) for nice bite sized transform examples, and read the extensive Shiftr javadoc.

Our unit tests follow the pattern :

{
    "input": {
        // sample input
    },
    "spec": {
        // transform spec
    },
    "expected": {
        // what the output of the transform looks like
    }
}

We read in "input", apply the "spec", and Diffy it against the "expected".

To learn the Shiftr DSL, examine "input" and "output" json, get an understanding of how data is moving, and then look at the transform spec to see how it facilitates the transform.

For reference, this was the very first test we wrote.

Demo

There is a demo available at jolt-demo.appspot.com. You can paste in JSON input data and a Spec, and it will post the data to server and run the transform.

Note

• it is hosted on a free Google App Engine instance, so it may take a minute to spin up.
• it validates in input JSON and spec client side.

Getting Started

Getting started code wise has its own doc.

Getting Transform Help

If you can't get a transform working and you need help, create and Issue in Jolt (for now).

Make sure you include what your "input" is, and what you want your "output" to be.

Why Jolt Exists

Aside from writing your own custom code to do a transform, there are two general approaches to doing a JSON to JSON transforms in Java.

1. JSON -> XML -> XSLT or STX -> XML -> JSON

Aside from being a Rube Goldberg approach, XSLT is more complicated than Jolt because it is trying to do the whole transform with a single DSL.

2. Write a Template (Velocity, FreeMarker, etc) that take hydrated JSON input and write textual JSON output

With this approach you are working from the output format backwards to the input, which is complex for any non-trivial transform. Eg, the structure of your template will be dictated by the output JSON format, and you will end up coding a parallel tree walk of the input data and the output format in your template. Jolt works forward from the input data to the output format which is simpler, and it does the parallel tree walk for you.

Alternatives

Being in the Java JSON processing "space", here are some other interesting JSON manipulation tools to look at / consider :

• jq - Awesome command line tool to extract data from JSON files (use it all the time, available via brew)
• JsonPath - Java : Extract data from JSON using XPATH like syntax.
• JsonSurfer - Java : Streaming JsonPath processor dedicated to processing big and complicated JSON data.

Performance

The primary goal of Jolt was to improve "developer speed" by providing the ability to have a declarative rather than imperative transforms. That said, Jolt should have a better runtime than the alternatives listed above.

Work has been done to make the stock Jolt transforms fast:

1. Transforms can be initialized once with their spec, and re-used many times in a multi-threaded environment.
   • We reuse initialized Jolt transforms to service multiple web requests from a DropWizard service.
2. "*" wildcard logic was redone to reduce the use of Regex in the common case, which was a dramatic speed improvement.
3. The parallel tree walk performed by Shiftr was optimized.

Two things to be aware of :

1. Jolt is not "stream" based, so if you have a very large Json document to transform you need to have enough memory to hold it.
2. The transform process will create and discard a lot of objects, so the garbage collector will have work to do.

Jolt CLI

Jolt Transforms and tools can be run from the command line. Command line interface doc here.

Code Coverage

For the moment we have Cobertura configured in our poms.

mvn cobertura:cobertura
open jolt-core/target/site/cobertura/index.html

Currently, for the jolt-core artifact, code coverage is at 89% line, and 83% branch.

Release Notes

Versions and Release Notes available here.

In order to get optimal performance from cassandra, its important to understand how it stores the data on disk. Its common problem among new users coming from RDBMS to not consider the queries while designing their column families(a.k.a tables). Cassandra’s cql interface return data in tabular format and it might give the illusion that we can query it just like any RDBMS, but that’s not the case.

All cassandra data is persisted in SSTables(Sorted String tables) inside data directory. Default location of data directory is $CASSANDRA_HOME/data/data. You can change it using data_file_directories config in conf/cassandra.yaml. On a fresh setup, here’s what data directory looks like:

1
2
3
4
5
6

data/
├── system
├── system_auth
├── system_distributed
├── system_schema
└── system_traces

Each directory in data represent a keyspace. These are internal keyspaces used by cassandra.

Lets create a new keyspace

1

CREATE KEYSPACE ks1 WITH replication={'class':'SimpleStrategy','replication_factor':1};

Since we have not yet created any table, so you will not see ks1 named directory yet in data dir, but you can check cassandra' system_schema.keyspace table.

1
2
3
4
5

cqlsh> select * from system_schema.keyspaces where keyspace_name = 'ks1';
 keyspace_name | durable_writes | replication
---------------+----------------+-------------------------------------------------------------------------------------
           ks1 |           True | {'class': 'org.apache.cassandra.locator.SimpleStrategy', 'replication_factor': '1'}

Let create a table now:

1
2
3
4
5
6
7
8

CREATE TABLE user_tracking (
  user_id text,
  action_category text,
  action_id text,
  action_detail text,
  PRIMARY KEY(user_id, action_category, action_id)
);

As soon as you create the table, here’s how the data directory will look like

1
2
3

ks1/
└── tb1-ed4784f0b64711e7b18a2f179b6f38f9
    └── backups

It created a directory by name of <table>-<table_id>. Cassandra creates a sub directory for each table. All the data for this table will be contained in this dir. You can move around this dir to different location in future by creating a symlink.

You can check for this table in system_schema.tables table

1
2
3
4
5
6
7

cqlsh:ks1> select * from system_schema.tables where keyspace_name = 'ks1';
 keyspace_name | table_name | bloom_filter_fp_chance | caching                                       | comment | compaction                                                                                                                | compression                                                                             | crc_check_chance | dclocal_read_repair_chance | default_time_to_live | extensions | flags        | gc_grace_seconds | id                                   | max_index_interval | memtable_flush_period_in_ms | min_index_interval | read_repair_chance | speculative_retry
---------------+------------+------------------------+-----------------------------------------------+---------+---------------------------------------------------------------------------------------------------------------------------+-----------------------------------------------------------------------------------------+------------------+----------------------------+----------------------+------------+--------------+------------------+--------------------------------------+--------------------+-----------------------------+--------------------+--------------------+-------------------
           ks1 |        tb1 |                   0.01 | {'keys': 'ALL', 'rows_per_partition': 'NONE'} |         | {'class': 'org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy', 'max_threshold': '32', 'min_threshold': '4'} | {'chunk_length_in_kb': '64', 'class': 'org.apache.cassandra.io.compress.LZ4Compressor'} |                1 |                        0.1 |                    0 |           {} | {'compound'} |           864000 | ed4784f0-b647-11e7-b18a-2f179b6f38f9 |               2048 |                           0 |                128 |                  0 |      99PERCENTILE
(1 rows)

You will notice the table defaults like gc_grace_seconds and memtable_flush_period_in_ms which got applied.

Lets insert some data to this table:

1

insert into ks1.user_tracking(user_id, action_category, action_id,  action_detail) VALUES ('user1', 'auth', 'a1',  'Logged in from home page');

Now lets check the table data directory. You will notice that there is no new file created. Thats because C* first writes data in memory and then flushed it to disk after a certain threshold is reached. For durability purpose it writes data to a append only commit log. Commit log is shared across keyspaces. By default its location is $CASSANDRA_HOME/data/commitlog and it can be changed in conf/cassandra.yaml

To force flushing memtables data to disk as SSTable, use following command

1
2
3

bin/nodetool flush  //flush all tables from of all keyspace
bin/nodetool flush <keyspace>    //flush all tables for a single keyspace 
bin/nodetool flush <keyspace> <table_name>  //flush a single table from a keyspace

After flushing, here’s what the content of table directory looks like. Each flush create a new SSTable on disk. For each SSTable, following set of files is created

1
2
3
4
5
6
7
8
9
10
11
12

user_tracking-49eb78d0b65a11e7b18a2f179b6f38f9/
├── backups
├── mc-5-big-CompressionInfo.db
├── mc-5-big-Data.db
├── mc-5-big-Digest.crc32
├── mc-5-big-Filter.db
├── mc-5-big-Index.db
├── mc-5-big-Statistics.db
├── mc-5-big-Summary.db
└── mc-5-big-TOC.txt
1 directory, 8 files

All files share common naming convention which is --.db

• version is an alphabetic string which represents SSTable storage format version. Like mc in above in example
• generation is an index number which is incremented every time a new SSTable is created for a table. Like 5 in above example
• component represents the type of information stored in the file. Like CompressionInfo, Data, Digest, Index, etc

The table below provides a brief description for each component.

1
2
3
4
5
6
7
8
9

File Description
mc-1-big-TOC.txt  A file that lists the components for the given SSTable.
mc-1-big-Digest.crc32 A file that consists of a checksum of the data file.
mc-1-big-CompressionInfo.db   A file that contains meta data for the compression algorithm, if enabled.
mc-1-big-Statistics.db    A file that holds statistical metadata about the SSTable.
mc-1-big-Index.db A file that contains the primary index data.
mc-1-big-Summary.db   This file provides summary data of the primary index, e.g. index boundaries, and is supposed to be stored in memory.
mc-1-big-Filter.db    This file embraces a data structure used to validate if row data exists in memory i.e. to minimize the access of data on disk.
mc-1-big-Data.db  This file contains the base data itself. Note: All the other component files can be regenerated from the base data file.

Lets insert few more entries into the table and see what data directory looks like:

1
2
3
4
5

insert into ks1.user_tracking(user_id, action_category, action_id,  action_detail) VALUES ('user1', 'auth', 'a1',  'Logged in from home page');
insert into ks1.user_tracking(user_id, action_category, action_id,  action_detail) VALUES ('user1', 'auth', 'a2', 'Logged in from email link');
insert into ks1.user_tracking(user_id, action_category, action_id,  action_detail) VALUES ('user1', 'dashboard', 'a3', 'Opened dashboard link');
insert into ks1.user_tracking(user_id, action_category, action_id,  action_detail) VALUES ('user2', 'auth', 'a4', 'Logged in');

Again do bin/nodetool flush to flush data to SSTable and check filesystem:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

sam@sam-ub:ks1$ tree user_tracking-49eb78d0b65a11e7b18a2f179b6f38f9/
user_tracking-49eb78d0b65a11e7b18a2f179b6f38f9/
├── backups
├── mc-7-big-CompressionInfo.db
├── mc-7-big-Data.db
├── mc-7-big-Digest.crc32
├── mc-7-big-Filter.db
├── mc-7-big-Index.db
├── mc-7-big-Statistics.db
├── mc-7-big-Summary.db
├── mc-7-big-TOC.txt
├── mc-8-big-CompressionInfo.db
├── mc-8-big-Data.db
├── mc-8-big-Digest.crc32
├── mc-8-big-Filter.db
├── mc-8-big-Index.db
├── mc-8-big-Statistics.db
├── mc-8-big-Summary.db
└── mc-8-big-TOC.txt
1 directory, 16 files

You will notice that a new SSTable with generation number 7 is created. Cassandra periodically keep merging these SSTables through a process called compaction. You can force compaction by running following command.

1
2
3

bin/nodetool compact  //compact SSTables for tables from of all keyspace
bin/nodetool compact <keyspace>    //compact SSTables for tables of a single keyspace
bin/nodetool compact <keyspace> <table_name>  //compact SSTables of a single table

You will notice that after compaction is complete there is just a single SSTable with a new generation. Compaction causes a temporary spike in disk space usage and disk I/O while old and new SSTables co-exist. As it completes, compaction frees up disk space occupied by old SSTables.

Now lets see how data is exactly stored in this SSTable.

C* comes bundled with a utility called sstabledump which can be used to see content of SSTable in json or row format. Here’s what you will see

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

[
  {
    "partition" : {
      "key" : [ "user2" ],
      "position" : 0
    },
    "rows" : [
      {
        "type" : "row",
        "position" : 45,
        "clustering" : [ "auth", "a4" ],
        "liveness_info" : { "tstamp" : "2017-10-24T20:33:32.772370Z" },
        "cells" : [
          { "name" : "action_detail", "value" : "Logged in" }
        ]
      }
    ]
  },
  {
    "partition" : {
      "key" : [ "user1" ],
      "position" : 46
    },
    "rows" : [
      {
        "type" : "row",
        "position" : 104,
        "clustering" : [ "auth", "a1" ],
        "liveness_info" : { "tstamp" : "2017-10-24T20:33:32.074848Z" },
        "cells" : [
          { "name" : "action_detail", "value" : "Logged in from home page" }
        ]
      },
      {
        "type" : "row",
        "position" : 104,
        "clustering" : [ "auth", "a2" ],
        "liveness_info" : { "tstamp" : "2017-10-24T20:33:32.085959Z" },
        "cells" : [
          { "name" : "action_detail", "value" : "Logged in from email link" }
        ]
      },
      {
        "type" : "row",
        "position" : 145,
        "clustering" : [ "dashboard", "a3" ],
        "liveness_info" : { "tstamp" : "2017-10-24T20:33:32.099739Z" },
        "cells" : [
          { "name" : "action_detail", "value" : "Opened dashboard link" }
        ]
      }
    ]
  }
]

Here you will notice 2 “rows”. Don’t confuse this row with RDBMS row. This “row” actually means a paritition. Number of rows(partitions) in sstable is determined by your primary key. Here’s the key which we used:

Primary key consists of 2 parts - partitioning and clustering fields

First part of our key, i.e. “user” will be used for partitioning, and remaining part - aka action_category and action_id will be used for clustering.

To understand this better, lets create a new table with exact same definition but changed partitioning key

1
2
3
4
5
6
7
8

CREATE TABLE user_tracking_new (
  user_id text,
  action_category text,
  action_id text,
  action_detail text,
  PRIMARY KEY((user_id, action_category), action_id)   
);

In above definition, now paritioning key will be user_id + action_category and clustering key will be just action_id. Lets insert exact same 2 rows and notice how they gets stored in sstable

1
2
3
4

insert into ks1.user_tracking_new(user_id, action_category, action_id,  action_detail) VALUES ('user1', 'auth', 'a1', 'Logged in from home page');
insert into ks1.user_tracking_new(user_id, action_category, action_id,  action_detail) VALUES ('user1', 'auth', 'a2', 'Logged in from email link');
insert into ks1.user_tracking_new(user_id, action_category, action_id,  action_detail) VALUES ('user1', 'dashboard', 'a3', 'Opened dashboard link');
insert into ks1.user_tracking_new(user_id, action_category, action_id,  action_detail) VALUES ('user2', 'auth', 'a4', 'Logged in');

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

[
  {
    "partition" : {
      "key" : [ "user1", "dashboard" ],
      "position" : 0
    },
    "rows" : [
      {
        "type" : "row",
        "position" : 67,
        "clustering" : [ "a3" ],
        "liveness_info" : { "tstamp" : "2017-10-24T20:32:45.633901Z" },
        "cells" : [
          { "name" : "action_detail", "value" : "Opened dashboard link" }
        ]
      }
    ]
  },
  {
    "partition" : {
      "key" : [ "user2", "auth" ],
      "position" : 68
    },
    "rows" : [
      {
        "type" : "row",
        "position" : 118,
        "clustering" : [ "a4" ],
        "liveness_info" : { "tstamp" : "2017-10-24T20:32:45.648367Z" },
        "cells" : [
          { "name" : "action_detail", "value" : "Logged in" }
        ]
      }
    ]
  },
  {
    "partition" : {
      "key" : [ "user1", "auth" ],
      "position" : 119
    },
    "rows" : [
      {
        "type" : "row",
        "position" : 182,
        "clustering" : [ "a1" ],
        "liveness_info" : { "tstamp" : "2017-10-24T20:32:45.614746Z" },
        "cells" : [
          { "name" : "action_detail", "value" : "Logged in from home page" }
        ]
      },
      {
        "type" : "row",
        "position" : 182,
        "clustering" : [ "a2" ],
        "liveness_info" : { "tstamp" : "2017-10-24T20:32:45.624710Z" },
        "cells" : [
          { "name" : "action_detail", "value" : "Logged in from email link" }
        ]
      }
    ]
  }
]

You will notice that in this example, for each user, each category of data went into a different partition.

First rule for data modelling is that you should choose your paritioning key in a way for a single query, only 1 partition is read.

Lets try to run some queries and see the impact of partitioning:

1
2
3
4
5
6
7
8
9

cqlsh:ks1> select * from user_tracking where user_id = 'user1';
 user_id | action_category | action_id | action_detail
---------+-----------------+-----------+---------------------------
   user1 |            auth |        a1 |  Logged in from home page
   user1 |            auth |        a2 | Logged in from email link
   user1 |       dashboard |        a3 |     Opened dashboard link
(3 rows)

Lets run same query on user_tracking_new

1
2

cqlsh:ks1> select * from user_tracking_new where user_id = 'user2';
InvalidRequest: Error from server: code=2200 [Invalid query] message="Partition key parts: action_category must be restricted as other parts are"

Reason you got the error is that at minimum you need to specify all columns which are part of partition key. C* needs to know a way to get to a parition before it can query data inside it. In first case it worked because only user_id was part of partition key.

That means our user_tracking_new can only be used to query user data when category is also know. You might wonder then why do you even prefer that over first CF. Reason is that in first CF, for huge volume of user active, parition can grow very large and hence have performance issues. Our goal is to keep partition size to a reasonable size to not effect query performance.

Lets try another query:

1
2
3

cqlsh:ks1> select * from user_tracking_new where user_id = 'user1' and action_category = 'auth' and action_detail = 'Logged in from home page';
InvalidRequest: Error from server: code=2200 [Invalid query] message="Cannot execute this query as it might involve data filtering and thus may have unpredictable performance. If you want to execute this query despite the performance unpredictability, use ALLOW FILTERING"

You will see that it failed to fetch the result. Reason being that above filtering needs to span 2 partitions and hence C* is warning against it. If we want to force it, then we can add ALLOW FILTERING at the end of query

1
2
3
4
5
6
7

cqlsh:ks1> select * from user_tracking_new where user_id = 'user1' and action_category = 'auth' and action_detail = 'Logged in from home page' ALLOW FILTERING;
 user_id | action_category | action_id | action_detail
---------+-----------------+-----------+--------------------------
   user1 |            auth |        a1 | Logged in from home page
(1 rows)

This is another extremely powerful feature available in Cassandra, and it allows you to naturally store records in a given order based on the value of a particular column. So every time you write to the Stocks table, Cassandra will figure out where that record is supposed to go in the physical data partitions and store the record in the order you told it to. Storing data in sorted order gives drastic query performance improvements for range queries, which is very significant in timeseries data.

1
2
3
4
5
6
7
8

CREATE TABLE user_tracking_ordered (
  user_id text,
  action_category text,
  action_id text,
  action_detail text,
  PRIMARY KEY((user_id, action_category), action_id)
) WITH CLUSTERING ORDER BY (action_id DESC); 

Now insert same data in this CF:

1
2
3
4
5

insert into ks1.user_tracking_ordered(user_id, action_category, action_id,  action_detail) VALUES ('user1', 'auth', 'a1', 'Logged in');
insert into ks1.user_tracking_ordered(user_id, action_category, action_id,  action_detail) VALUES ('user1', 'auth', 'a2', 'Logged in');
insert into ks1.user_tracking_ordered(user_id, action_category, action_id,  action_detail) VALUES ('user1', 'auth', 'a3', 'Logged in');
insert into ks1.user_tracking_ordered(user_id, action_category, action_id,  action_detail) VALUES ('user1', 'auth', 'a4', 'Logged in');
insert into ks1.user_tracking_ordered(user_id, action_category, action_id,  action_detail) VALUES ('user1', 'auth', 'a5', 'Logged in');

If you try to fetch 1 row, you will get it in sorted order

1
2
3
4
5
6
7

cqlsh:ks1> select * from user_tracking_ordered limit 1;
 user_id | action_category | action_id | action_detail
---------+-----------------+-----------+---------------
   user1 |            auth |        a1 |     Logged in
(1 rows)

If you change CLUSTERING order to DESC like WITH CLUSTERING ORDER BY (action_id DESC) and do same thing, you can see it now return it sorted in DESC order:

1
2
3
4
5
6
7

cqlsh:ks1> select * from user_tracking_ordered limit 1;
 user_id | action_category | action_id | action_detail
---------+-----------------+-----------+---------------
   user1 |            auth |        a5 |     Logged in
(1 rows)

You can see same ordered getting reflected in sstable too

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

$ sstabledump mc-*-big-Data.db
[
  {
    "partition" : {
      "key" : [ "user1", "auth" ],
      "position" : 0
    },
    "rows" : [
      {
        "type" : "row",
        "position" : 50,
        "clustering" : [ "a5" ],
        "liveness_info" : { "tstamp" : "2017-10-25T08:21:43.628921Z" },
        "cells" : [
          { "name" : "action_detail", "value" : "Logged in" }
        ]
      },
      {
        ...
        "clustering" : [ "a4" ],
        ...
      },
      {
        ...
        "clustering" : [ "a3" ],
        ...
      },
      {
        ...
        "clustering" : [ "a2" ],
        ...
      },
      {
          ...
        "clustering" : [ "a1" ],
        ...
      }
    ]
  }
]

You will notice that for each column stored in cassandra, it has to maintain some other metadata too. You need to understand these overheads too for better capacity planning. I will cover those in some future post.

I hope you got some idea about why cqlsh looks like SQL but doesn’t behave like one. Play around with different combinations of primary key and see what data looks like in SSTable to get more understanding of data storage which will help you to model it based on your queries.

Hundreds of benchmark network data sets

Download hundreds of benchmark network data sets from a variety of network types. Also share and contribute by uploading recent network data sets. Naturally all conceivable data may be represented as a graph for analysis. This includes social network data, brain networks, temporal network data, web graph datasets, road networks, retweet networks, labeled graphs, and numerous other real-world graph datasets.

Network data can be visualized and explored in real-time on the web via our web-based interactive network visual analytics platform.

Dagster is a data orchestrator for machine learning, analytics, and ETL

Dagster lets you define pipelines in terms of the data flow between reusable, logical components, then test locally and run anywhere. With a unified view of pipelines and the assets they produce, Dagster can schedule and orchestrate Pandas, Spark, SQL, or anything else that Python can invoke.

Dagster is designed for data platform engineers, data engineers, and full-stack data scientists. Building a data platform with Dagster makes your stakeholders more independent and your systems more robust. Developing data pipelines with Dagster makes testing easier and deploying faster.

Develop and test on your laptop, deploy anywhere

With Dagster’s pluggable execution, the same pipeline can run in-process against your local file system, or on a distributed work queue against your production data lake. You can set up Dagster’s web interface in a minute on your laptop, or deploy it on-premise or in any cloud.

Model and type the data produced and consumed by each step

Dagster models data dependencies between steps in your orchestration graph and handles passing data between them. Optional typing on inputs and outputs helps catch bugs early.

Link data to computations

Dagster’s Asset Manager tracks the data sets and ML models produced by your pipelines, so you can understand how your they were generated and trace issues when they don’t look how you expect.

Build a self-service data platform

Dagster helps platform teams build systems for data practitioners. Pipelines are built from shared, reusable, configurable data processing and infrastructure components. Dagster’s web interface lets anyone inspect these objects and discover how to use them.

Avoid dependency nightmares

Dagster’s repository model lets you isolate codebases, so that problems in one pipeline don’t bring down the rest. Each pipeline can have its own package dependencies and Python version. Pipelines run in isolated processes so user code issues can't bring the system down.

Debug pipelines from a rich UI

Dagit, Dagster’s web interface, includes expansive facilities for understanding the pipelines it orchestrates. When inspecting a pipeline run, you can query over logs, discover the most time consuming tasks via a Gantt chart, re-execute subsets of steps, and more.

Getting Started

Installation

pip install dagster dagit

This installs two modules:

• Dagster: the core programming model and abstraction stack; stateless, single-node, single-process and multi-process execution engines; and a CLI tool for driving those engines.
• Dagit: the UI for developing and operating Dagster pipelines, including a DAG browser, a type-aware config editor, and a live execution interface.

Hello dagster 👋

hello_dagster.py

from dagster import execute_pipeline, pipeline, solid
@solid
def get_name(_):
    return 'dagster'
@solid
def hello(context, name: str):
    context.log.info('Hello, {name}!'.format(name=name))
@pipeline
def hello_pipeline():
    hello(get_name())

Save the code above in a file named hello_dagster.py. You can execute the pipeline using any one of the following methods:

(1) Dagster Python API

if __name__ == "__main__":
    execute_pipeline(hello_pipeline)   # Hello, dagster!

(2) Dagster CLI

$ dagster pipeline execute -f hello_dagster.py

(3) Dagit web UI

$ dagit -f hello_dagster.py

Learn

Next, jump right into our tutorial, or read our complete documentation. If you're actively using Dagster or have questions on getting started, we'd love to hear from you:

Contributing

For details on contributing or running the project for development, check out our contributing guide.

Integrations

Dagster works with the tools and systems that you're already using with your data, including:

Integration		Dagster Library
	Apache Airflow	dagster-airflow Allows Dagster pipelines to be scheduled and executed, either containerized or uncontainerized, as Apache Airflow DAGs.
	Apache Spark	dagster-spark · dagster-pyspark Libraries for interacting with Apache Spark and PySpark.
	Dask	dagster-dask Provides a Dagster integration with Dask / Dask.Distributed.
	Datadog	dagster-datadog Provides a Dagster resource for publishing metrics to Datadog.
/	Jupyter / Papermill	dagstermill Built on the papermill library, dagstermill is meant for integrating productionized Jupyter notebooks into dagster pipelines.
	PagerDuty	dagster-pagerduty A library for creating PagerDuty alerts from Dagster workflows.
	Snowflake	dagster-snowflake A library for interacting with the Snowflake Data Warehouse.
Cloud Providers
	AWS	dagster-aws A library for interacting with Amazon Web Services. Provides integrations with Cloudwatch, S3, EMR, and Redshift.
	Azure	dagster-azure A library for interacting with Microsoft Azure.
	GCP	dagster-gcp A library for interacting with Google Cloud Platform. Provides integrations with GCS, BigQuery, and Cloud Dataproc.

This list is growing as we are actively building more integrations, and we welcome contributions!

Big Data is complex, I have written quite a bit about the vast ecosystem and the wide range of options available. One aspect that is often ignored but critical, is managing the execution of the different steps of a big data pipeline. Quite often the decision of the framework or the design of the execution process is deffered to a later stage causing many issues and delays on the project.

You should design your pipeline orchestration early on to avoid issues during the deployment stage. Orchestration should be treated like any other deliverable; it should be planned, implemented, tested and reviewed by all stakeholders.

Orchestration frameworks are often ignored and many companies end up implementing custom solutions for their pipelines. This is not only costly but also inefficient, since custom orchestration solutions tend to face the same problems that out-of-the-box frameworks already have solved; creating a long cycle of trial and error.

In this article, I will present some of the most common open source orchestration frameworks.

Data pipeline orchestration is a cross cutting process which manages the dependencies between your pipeline tasks, schedules jobs and much more. If you use stream processing, you need to orchestrate the dependencies of each streaming app, for batch, you need to schedule and orchestrate the jobs.

Remember, tasks and applications may fail, so you need a way to schedule, reschedule, replay, monitor, retry and debug your whole data pipeline in an unified way.

Some of the functionality provided by orchestration frameworks are:

• Job Scheduling
• Dependency management
• Error management and retries
• Job parametrization
• SLAs tracking, alerting and notification
• UI with dashboards such Gantt charts and graphs
• History and audit
• Data storage for metadata
• Log aggregation

Let’s review some of the options…

Apache Oozie it’s a scheduler for Hadoop, jobs are created as DAGs and can be triggered by a cron based schedule or data availability. Oozie is a scalable, reliable and extensible system that runs as a Java web application. It has integrations with ingestion tools such as Sqoop and processing frameworks such Spark.

Oozie workflows definitions are written in hPDL (XML). Workflows contain control flow nodes and action nodes. Control flow nodes define the beginning and the end of a workflow ( start, end and fail nodes) and provide a mechanism to control the workflow execution path ( decision, fork and join nodes)[1].

Action nodes are the mechanism by which a workflow triggers the execution of a task. Oozie provides support for different types of actions (map-reduce, Pig, SSH, HTTP, eMail…) and can be extended to support additional type of actions[1].

Also, workflows can be parameterized and several identical workflow jobs can concurrently.

It was the first scheduler for Hadoop and quite popular but has become a bit outdated, still is a great choice if you rely entirely in the Hadoop platform.

Airflow is a platform that allows to schedule, run and monitor workflows. It has become the most famous orchestrator for big data pipelines thanks to the ease of use and the innovate workflow as code approach where DAGs are defined in Python code that can be tested as any other software deliverable.

It uses DAGs to create complex workflows. Each node in the graph is a task, and edges define dependencies among the tasks. Tasks belong to two categories:

• Operators: execute some operation.
• Sensors: check for the state of a process or a data structure.

Airflow scheduler executes your tasks on an array of workers while following the specified dependencies described by you. It has a modular architecture and uses a message queue to orchestrate an arbitrary number of workers and can scale to infinity[2].

It generates the DAG for you, maximizing parallelism. The DAGs are written in Python, so you can run them locally, unit test them and integrate them with your development workflow. When workflows are defined as code, they become more maintainable, versionable, testable, and collaborative[2].

The rich UI makes it easy to visualize pipelines running in production, monitor progress, and troubleshoot issues when needed[2]. It is fast, easy to use and very useful. It has several views and many ways to troubleshoot issues. It keeps the history of your runs for later reference.

It is very straightforward to install. You just need Python. It has two processes, the UI and the Scheduler that run independently.

Principles[2]:

• Dynamic: Airflow pipelines are configuration as code (Python), allowing for dynamic pipeline generation. This allows for writing code that instantiates pipelines dynamically.
• Extensible: Easily define your own operators, executors and extend the library so that it fits the level of abstraction that suits your environment.
• Elegant: Airflow pipelines are lean and explicit. Parameterizing your scripts is built into Airflow using the powerful Jinja templating engine.
• Scalable

Although Airflow flows are written as code, Airflow is not a data streaming solution[2]. Also, workflows are expected to be mostly static or slowly changing, for very small dynamic jobs there are other options that we will discuss later.

It is simple and stateless, although XCOM functionality is used to pass small metadata between tasks which is often required, for example when you need some kind of correlation ID. It also supports variables and parameterized jobs. Finally, it has support SLAs and alerting. It can be integrated with on-call tools for monitoring.

Luigi is an alternative to Airflow with similar functionality but Airflow has more functionality and scales up better than Luigi.

Dagster is a newer orchestrator for machine learning, analytics, and ETL[3]. The main difference is that you can track the inputs and outputs of the data, similar to Apache NiFi, creating a data flow solution. This mean that it tracks the execution state and can materialize values as part of the execution steps. You can test locally and run anywhere with a unified view of data pipelines and assets. It support any cloud environment.

Dagster models data dependencies between steps in your orchestration graph and handles passing data between them. Optional typing on inputs and outputs helps catch bugs early[3]. Pipelines are built from shared, reusable, configurable data processing and infrastructure components. Dagster’s web UI lets anyone inspect these objects and discover how to use them[3].

It can also run several jobs in parallel, it is easy to add parameters, easy to test, provides simple versioning, great logging, troubleshooting capabilities and much more. It is more feature rich than Airflow but it is still a bit immature and due to the fact that it needs to keep track the data, it may be difficult to scale, which is a problem shared with NiFi due to the stateful nature. Also it is heavily based on the Python ecosystem.

Prefect is similar to Dagster, provides local testing, versioning, parameter management and much more. It is also Python based.

What makes Prefect different from the rest is that aims to overcome the limitations of Airflow execution engine such as improved scheduler, parametrized workflows, dynamic workflows, versioning and improved testing. Versioning is a must have for many DevOps oriented organizations which is still not supported by Airflow and Prefect does support it.

It has a core open source workflow management system and also a cloud offering which requires no setup at all. Prefect Cloud is powered by GraphQL, Dask, and Kubernetes, so it’s ready for anything[4]. The UI is only available in the cloud offering.

Apache NiFi is not an orchestration framework but a wider dataflow solution. NiFi can also schedule jobs, monitor, route data, alert and much more. It is focused on data flow but you can also process batches.

It does not require any type of programming and provides a drag and drop UI. It is very easy to use and you can use it for easy to medium jobs without any issues but it tends to have scalability problems for bigger jobs.

It runs outside of Hadoop but can trigger Spark jobs and connect to HDFS/S3.

Let’s put see some examples…

• I have a legacy Hadoop cluster with slow moving Spark batch jobs, your team is conform of Scala developers and your DAG is not too complex. In this case, Ozzie is a good choice since it provides a simple way to schedule Spark jobs.
• I have many slow moving Spark jobs with complex dependencies, you need to be able to test the dependencies and maximize parallelism, you want a solution that is easy to deploy and provides lots of troubleshooting capabilities. In this case, Airflow is your best bet.
• I need to ingest data in real time from many sources, you need to track the data lineage, route the data, enrich it and be able to debug any issues. This is a real time data streaming pipeline required by your BAs which do not have much programming knowledge. In this case, Apache NiFi is your best option since provides all the features required without the need of Python Skills. If your team poses Python skills, consider Dagster.
• I would like to create real time and batch pipelines in the cloud without having to worried about maintaining servers or configuring system. I need a quick, powerful solution to empower my Python based analytics team. In this case, use Prefect Cloud.
• I have short lived, fast moving jobs which deal with complex data that I would like to track, I need a way to troubleshoot issues and make changes in quick in production. In this case consider Dagster.
• I deal with hundreds of terabytes of data, I have a complex dependencies and I would like to automate my workflow tests. For this case, use Airflow since it can scale, interact with many system and can be unit tested. Dagster or Prefect may have scale issue with data at this scale.
• I’m not sure about what I need. In this case, start with Airflow since it is the most popular choice.

We have seem some of the most common orchestration frameworks. As you can see, most of them use DAGs as code so you can test locally, debug pipelines and test them properly before rolling new workflows to production. Consider all the features discussed in this article and choose the best tool for the job.

In short, if your requirement is just orchestrate independent tasks that do not require to share data and/or you have slow jobs and/or you do not use Python, use Airflow or Ozzie. For data flow applications that require data lineage and tracking use NiFi for non developers; or Dagster or Prefect for Python developers.

When possible, try to keep jobs simple and manage the data dependencies outside the orchestrator, this is very common in Spark where you save the data to deep storage and not pass it around. In this case, Airflow is a great option since it doesn’t need to track the data flow and you can still pass small meta data like the location of the data using XCOM. For smaller, faster moving , python based jobs or more dynamic data sets, you may want to track the data dependencies in the orchestrator and use tools such Dagster.

[1] https://oozie.apache.org/docs/5.2.0/index.html

[2] https://airflow.apache.org/docs/stable/

[3] https://docs.dagster.io/

[4] https://docs.prefect.io/

[5] https://nifi.apache.org/