Use in prod

scripts/create-cluster-bankfacil.sh

@@ -0,0 +1,74 @@
+#!/usr/bin/env bash
+
+set -eu
+
+APP_NAME="topic-reloader"
+
+
+readonly DEBUG=${DEBUG:-"no"}
+readonly ID=$(date -u +'%Y%m%d%M%S')
+readonly configurationFile="/tmp/${APP-NAME}-configurations-${ID}.json"
+readonly ec2AttributesFile="/tmp/${APP-NAME}-ec2Attributes-${ID}.json"
+readonly instanceGroupsFile="/tmp/${APP-NAME}-instanceGroups-${ID}.json"
+
+cat <<EOF >"${configurationFile}"
+[
+  {
+    "Classification": "spark-defaults",
+    "Properties": {
+      "spark.jars.packages": "org.apache.hadoop:hadoop-aws:3.0.0",
+      "spark.serializer": "org.apache.spark.serializer.KryoSerializer"
+    }
+  },
+  {
+    "Classification": "spark",
+    "Properties": {
+      "maximizeResourceAllocation": "true"
+    }
+  }
+]
+EOF
+
+cat <<EOF >"${ec2AttributesFile}"
+{
+  "KeyName": "mdm_emr",
+  "InstanceProfile": "EMR_EC2_DefaultRole",
+  "SubnetId": "subnet-fd58938b",
+  "EmrManagedSlaveSecurityGroup": "sg-1741836d",
+  "EmrManagedMasterSecurityGroup": "sg-fe468484"
+}
+EOF
+
+cat <<EOF >"${instanceGroupsFile}"
+[
+  {
+    "Name":"TopicReloaderMaster",
+    "InstanceCount": 1,
+    "InstanceGroupType": "MASTER",
+    "InstanceType": "r5.xlarge"
+  },
+  {
+    "Name": "TopicReloaderCore",
+    "InstanceCount": 2,
+    "InstanceGroupType": "CORE",
+    "InstanceType": "r5.xlarge"
+  }
+]
+EOF
+
+readonly me="$(whoami)"
+readonly name="${APP_NAME}-${ID}-${me}"
+
+[[ ${DEBUG} != "no" ]] && set -x
+
+aws emr create-cluster \
+  --name "${name}" \
+  --applications '[ { "Name": "Hadoop" }, { "Name": "Spark" }, { "Name": "Ganglia" } ]' \
+  --release-label "emr-6.1.0" \
+  --service-role "EMR_DefaultRole" \
+  --ec2-attributes "file://${ec2AttributesFile}" \
+  --instance-groups "file://${instanceGroupsFile}" \
+  --configurations "file://${configurationFile}" \
+  --log-uri "s3://prod-creditas-data-lake/${APP_NAME}/log" \
+  --enable-debugging \
+  --tags Name="${name}" Application="${name}" Environment="dev" Owner="${me}" Squad="Data Platform" | jq

scripts/run-job-bankfacil.sh

@@ -0,0 +1,95 @@
+#!/usr/bin/env bash
+
+set -eu
+
+APP_NAME="topic-reloader"
+
+readonly DEBUG=${DEBUG:-no}
+readonly ID=$(date -u +'%Y%m%d%M%S')
+readonly S3_JAR_DIR="s3://prod-creditas-data-lake/${APP_NAME}/jar"
+readonly HDFS_BASEDIR='/home/hadoop'
+
+function usage() {
+  cat <<USAGE
+help!
+USAGE
+}
+
+#############################
+# begin command line parser #
+#############################
+cluster_id=""
+jar=""
+topic_name=""
+usage="false"
+unknown=("")
+for param in "${@}"; do
+  case ${param} in
+  --cluster-id=*) cluster_id=${param#*=} ;;
+  --jar=*) jar=${param#*=} ;;
+  --topic-name=*) topic_name=${param#*=} ;;
+  help | -h | --help) usage="true" ;;
+  *) unknown+=("${param}") ;;
+  esac
+  shift
+done
+
+[[ ${#unknown} -gt 1 ]] && echo "unknown parameters: ${unknown[*]}" && exit 1
+[[ ${#unknown} -gt 0 ]] && echo "unknown parameter: ${unknown[*]}" && exit 1
+[[ ${usage} == true ]] && usage && exit 0
+
+errors=0
+[[ ${cluster_id} == "" ]] && errors=$((errors + 1)) && echo '--cluster-id=<String> must be provided'
+[[ ${jar} == "" ]] && errors=$((errors + 1)) && echo '--jar=<Path> must be provided'
+[[ ${topic_name} == "" ]] && errors=$((errors + 1)) && echo '--topic-name=<String> must be provided'
+
+[[ ${errors} -ne 0 ]] && printf '\n-> Please, fix above errors and try again, %d errors(s) found\n' ${errors} && exit 1
+###########################
+# end command line parser #
+###########################
+
+[[ ${DEBUG} != 'no' ]] && set -x
+
+readonly CLUSTER_ID=${cluster_id}
+readonly JAR_PATH=${jar}
+readonly HDFS_PATH="${HDFS_BASEDIR}/${ID}"
+readonly JAR=$(basename "${JAR_PATH}")
+
+readonly stepsFile="/tmp/${APP_NAME}-steps-${ID}.json"
+
+
+cat <<EOF >"${stepsFile}"
+[
+  {
+    "Type": "CUSTOM_JAR",
+    "ActionOnFailure": "CONTINUE",
+    "Name": "copy ${ID}",
+    "Jar": "command-runner.jar",
+    "Args": [
+      "s3-dist-cp",
+      "--s3Endpoint=s3.amazonaws.com",
+      "--src=${S3_JAR_DIR}",
+      "--dest=hdfs://${HDFS_PATH}"
+    ]
+  },
+  {
+    "Type": "Spark",
+    "ActionOnFailure": "CONTINUE",
+    "Name": "run ${ID}",
+    "Args": [
+      "--class",
+      "topic.reloader.main.TopicReloaderApp",
+      "hdfs://${HDFS_PATH}/${JAR}",
+      "${ID}",
+      "prod",
+      "${topic_name}"
+    ]
+  }
+]
+EOF
+
+aws s3 cp "${JAR_PATH}" "${S3_JAR_DIR}/"
+
+aws emr add-steps \
+  --cluster-id "${CLUSTER_ID}" \
+  --steps "file://${stepsFile}" | jq

src/main/resources/application.conf

@@ -3,3 +3,9 @@ dev.kafka-bootstrap-servers="pkc-ep9mm.us-east-2.aws.confluent.cloud:9092"
 dev.schema-registry-address="https://psrc-l7y22.us-east-2.aws.confluent.cloud"
 dev.kafka-api-key="IN4FE66KINLQI3RK"
 dev.schema-registry-api-key="YAG3SYAH56HDYGQA"
+
+prod.input-path-pattern="s3://prod-creditas-data-lake/scope=events/domain=DOMAIN/subdomain=SUBDOMAIN/processing-level=raw/event-name=EVENT_NAME"
+prod.kafka-bootstrap-servers="pkc-epjvm.us-east-1.aws.confluent.cloud:9092"
+prod.schema-registry-address="https://psrc-l6oz3.us-east-2.aws.confluent.cloud"
+prod.kafka-api-key="OXX3P7BBBEFRASGA"
+prod.schema-registry-api-key="6P5AN5XFFYHBXS6H"

src/main/scala/topic/reloader/main/TopicReloaderApp.scala

@@ -7,6 +7,8 @@ import org.apache.spark.sql.functions.struct
 import topic.reloader.WrappedSession
 import topic.reloader.lib.model.AppName
 import za.co.absa.abris.avro.functions.to_avro
+import za.co.absa.abris.avro.read.confluent.SchemaManagerFactory
+import za.co.absa.abris.avro.registry.{LatestVersion, SchemaSubject}
 import za.co.absa.abris.config.{AbrisConfig, ToAvroConfig}
 
 object TopicReloaderApp extends App {
@@ -43,10 +45,14 @@ object TopicReloaderApp extends App {
   val registryConfig = Map(
     AbrisConfig.SCHEMA_REGISTRY_URL -> schemaRegistryAddr,
     "basic.auth.credentials.source" -> "USER_INFO",
-    "basic.auth.user.info" -> (conf.getString("schema-registry-api-key")+":"+SR_PASSWORD)
+    "basic.auth.user.info" -> (conf.getString("schema-registry-api-key") + ":" + SR_PASSWORD)
   )
+  val schemaManager = SchemaManagerFactory.create(registryConfig)
+
 
   val reloadedTopicName = s"$topicName-reloaded"
+  val subject = SchemaSubject.usingTopicNameStrategy(reloadedTopicName)
+  val latestSchema = schemaManager.getSchemaBySubjectAndVersion(subject, LatestVersion())
 
   val toAvroConfig1: ToAvroConfig = AbrisConfig
     .toConfluentAvro
@@ -56,19 +62,27 @@ object TopicReloaderApp extends App {
 
   val sourceDF = spark.read
     .format("avro")
-    .option("mergeSchema", "true")
+    .option("avroSchema", latestSchema.toString)
     .load(inputDir)
     .drop("version", "year", "month", "day") //partitioning columns, they're not on the original event.
 
+  sourceDF.printSchema()
+  sourceDF.show()
+  println(s"src count: ${sourceDF.count()}")
+
   val allColumns = struct(sourceDF.columns.head, sourceDF.columns.tail: _*)
   val avroDF = sourceDF.select(to_avro(allColumns, toAvroConfig1) as 'value)
 
+  avroDF.printSchema()
+  avroDF.show()
+  println(s"dst count: ${avroDF.count()}")
+
   avroDF.write
     .format("kafka")
     .option("topic", reloadedTopicName)
     .option("kafka.bootstrap.servers", conf.getString("kafka-bootstrap-servers"))
     .option("kafka.security.protocol", "SASL_SSL")
-    .option("kafka.sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\""+conf.getString("kafka-api-key")+"\" password=\""+KAFKA_PASSWORD+"\";")
+    .option("kafka.sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"" + conf.getString("kafka-api-key") + "\" password=\"" + KAFKA_PASSWORD + "\";")
     .option("kafka.sasl.mechanism", "PLAIN")
     .save()
 }