BIGTOP-4370. Integrate the bigpetstore-spark workflow with Airflow.

bigtop-bigpetstore/bigpetstore-spark/README.md

@@ -173,3 +173,48 @@ spark-submit --master local[2] --class org.apache.bigtop.bigpetstore.spark.analy
 ```
 
 The resulting json file will contain lists of customers, products, and products recommended to each customer.
+
+Airflow Integration
+--------------------------------------------
+
+The steps described above are consolidated into [a single Airflow DAG](../../bigtop-deploy/puppet/modules/airflow/templates/example_bigpetstore.py).
+You can try it as follows. The example here is tested on Debian 12, and assuming [Puppet](../../bigtop_toolchain/bin/puppetize.sh) and [Bigtop toolchain](../../bigtop_toolchain/README.md) are already installed and BPS Spark is built as a shadowed JAR in accordance with [the guide above](#building-and-running-with-spark).
+
+1. (Optional) build Airflow and Spark including their dependencies.
+   You can skip it if you use Bigtop's binary distribution packages on its public repository.
+
+```
+$ ./gradlew allclean airflow-pkg spark-pkg repo -Dbuildwithdeps=true
+```
+
+2. Deploy the packages above though Bigtop's Puppet manifests with appropriate parameters:
+
+```
+$ cat bigtop-deploy/puppet/hieradata/site.yaml
+bigtop::bigtop_repo_gpg_check: false
+bigtop::bigtop_repo_uri: [...]
+bigtop::hadoop_head_node: ...
+hadoop::hadoop_storage_dirs: [/data]
+hadoop_cluster_node::cluster_components: [bigtop-utils, hdfs, yarn, spark, airflow]
+airflow::server::install_bigpetstore_example: true  # Enable the BigPetStore DAG
+airflow::server::load_examples: false  # Disable Airflow's default examples for simplicity
+$ sudo cp -r bigtop-deploy/puppet/hiera* /etc/puppet
+$ sudo puppet apply --hiera_config=/etc/puppet/hiera.yaml --modulepath=/vagrant_data/bigtop/bigtop-deploy/puppet/modules:/etc/puppet/code/modules:/usr/share/puppet/modules /vagrant_data/bigtop/bigtop-deploy/puppet/manifests
+```
+
+3. Create output directories on HDFS with the airflow owner:
+
+```
+$ sudo -u hdfs hdfs dfs -mkdir /user/airflow
+$ sudo -u hdfs hdfs dfs -chown airflow:airflow /user/airflow
+```
+
+4. Login Airflow's web UI with admin/admin and wait for the DAG picked up for a while.
+   Once it's found, you can run it through the triangle button on the right.
+   ![dag_list](images/dag_list.png)
+
+5. Trigger the DAG with the path to the BPS Spark JAR.
+   ![trigger_dag](images/trigger_dag.png)
+
+6. If settings are appropriate, the BPS DAG should successfully run as follows:
+   ![running_dag](images/running_dag.png)

bigtop-deploy/puppet/hieradata/bigtop/cluster.yaml

@@ -210,6 +210,11 @@ flink::common::parallelism_default: "1"
 flink::common::jobmanager_failover_strategy: "region"
 flink::common::rest_port: "8081"
 
-
 # Ranger
 ranger::admin::admin_password: "Admin01234"
+
+# Airflow
+airflow::server::executor: "SequentialExecutor"
+airflow::server::load_examples: "True"
+airflow::server::sql_alchemy_conn: "sqlite:////var/lib/airflow/airflow.db"
+airflow::server::install_bigpetstore_example: "False"

bigtop-deploy/puppet/manifests/cluster.pp

@@ -121,7 +121,10 @@
   },
   ranger => {
     master => ["ranger-server"],
-  }
+  },
+  airflow => {
+    master => ["airflow"],
+  },
 }
 
 class hadoop_cluster_node (
@@ -177,6 +180,7 @@
     "bigtop_utils",
     "phoenix",
     "ranger",
+    "airflow",
   ]
 
   node_with_roles::deploy_module { $modules:

bigtop-deploy/puppet/modules/airflow/manifests/init.pp

@@ -0,0 +1,82 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+class airflow {
+  class deploy ($roles) {
+    if ('airflow' in $roles) {
+      include airflow::server
+    }
+  }
+
+  class server($executor, $load_examples, $sql_alchemy_conn, $install_bigpetstore_example=False) {
+    package { 'airflow':
+      ensure => latest,
+    }
+
+    file { '/var/lib/airflow/airflow.cfg':
+      content => template('airflow/airflow.cfg'),
+      owner   => 'airflow',
+      group   => 'airflow',
+      require => Package['airflow'],
+    }
+
+    exec { 'airflow-db-init':
+      command     => '/usr/lib/airflow/bin/airflow db init',
+      environment => ['AIRFLOW_HOME=/var/lib/airflow'],
+      user        => 'airflow',
+      require     => File['/var/lib/airflow/airflow.cfg'],
+    }
+
+    exec { 'airflow-users-create':
+      command     => '/usr/lib/airflow/bin/airflow users create -e [email protected] -f John -l Doe -p admin -r Admin -u admin',
+      environment => ['AIRFLOW_HOME=/var/lib/airflow'],
+      user        => 'airflow',
+      require     => Exec['airflow-db-init'],
+    }
+
+    service { 'airflow-scheduler':
+      ensure  => running,
+      require => Exec['airflow-db-init'],
+    }
+
+    service { 'airflow-webserver':
+      ensure  => running,
+      require => Exec['airflow-db-init'],
+    }
+
+    if $install_bigpetstore_example {
+      exec { 'install-spark-provider':
+        command     => "/usr/lib/airflow/bin/python3 -m pip install apache-airflow-providers-apache-spark 'pyspark<4'",
+        environment => ['AIRFLOW_HOME=/var/lib/airflow'],
+        user        => 'root',
+        require     => Package['airflow'],
+      }
+
+      file { '/var/lib/airflow/dags':
+        ensure  => 'directory',
+        owner   => 'airflow',
+        group   => 'airflow',
+        require => Package['airflow'],
+      }
+
+      file { '/var/lib/airflow/dags/example_bigpetstore.py':
+        content => template('airflow/example_bigpetstore.py'),
+        owner   => 'airflow',
+        group   => 'airflow',
+        require => File['/var/lib/airflow/dags'],
+      }
+    }
+  }
+}