poststart.sh

#!/bin/bash -e
# Background process that copies extracts to gcsfuse mount on airflow-workers
#
# Author: Debnil Sur
# Date: October 2020
#
# This script is a long running background process on the airflow-worker
# nodes. It listens for the existing of new files in the /home/airflow/etlData
# directory, which is where data is exported using the stellar-etl framework.
# If the file is complete and closed, the script moves the file to the correct
# gcsfused mount, meaning that it is copied across all workers so that downstream
# Airflow tasks can reference the file.
#
# Follow the directions in the README.md to get the script mapped and executing
# on the airflow-worker nodes.

move_file_if_closed() {
    preFileSize=$(wc -c <"$1")
    sleep 10
    postFileSize=$(wc -c <"$1")
    lsof "$1"
    # if lsof returns an error code of 1, it means the file is not opened by any
    # other processes, allowing us to move it safely. Confirm there is data first.
    if [[ $? == 1 && $preFileSize -gt 0 ]]; then
        # File size is measured 10 seconds apart to confirm any batch process finished
        # writing to the file. If the sizes do not match, the file is still in use.
        if [[ $preFileSize == $postFileSize ]]; then
            mv "$1" /home/airflow/gcs/data/
        fi
    fi
}

sudo apt-get update && sudo apt-get install lsof
while true; do
    for file in /home/airflow/etlData/*; do
        if [[ -f "$file" ]]; then
            move_file_if_closed "$file"
        else
            for changeFile in "$file"/*; do
                if [[ -f "$changeFile" ]]; then
                    move_file_if_closed "$changeFile"
                fi
            done
        fi
    done
    sleep 1
done >/dev/null 2>&1 &
disown
disown -a
exit 0