From e3b20b466d58b3b12f21282bc74173a66a11a075 Mon Sep 17 00:00:00 2001 From: Eder Ruiz Date: Fri, 4 Dec 2020 11:48:27 -0300 Subject: [PATCH 1/6] Add scripts to bankfacil aws account (prod) --- scripts/create-cluster-bankfacil.sh | 74 ++++++++++++++++++++++ scripts/run-job-bankfacil.sh | 95 +++++++++++++++++++++++++++++ 2 files changed, 169 insertions(+) create mode 100755 scripts/create-cluster-bankfacil.sh create mode 100755 scripts/run-job-bankfacil.sh diff --git a/scripts/create-cluster-bankfacil.sh b/scripts/create-cluster-bankfacil.sh new file mode 100755 index 0000000..9b3a88f --- /dev/null +++ b/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 <"${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 <"${ec2AttributesFile}" +{ + "KeyName": "mdm_emr", + "InstanceProfile": "EMR_EC2_DefaultRole", + "SubnetId": "subnet-fd58938b", + "EmrManagedSlaveSecurityGroup": "sg-1741836d", + "EmrManagedMasterSecurityGroup": "sg-fe468484" +} +EOF + +cat <"${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 diff --git a/scripts/run-job-bankfacil.sh b/scripts/run-job-bankfacil.sh new file mode 100755 index 0000000..0ab7789 --- /dev/null +++ b/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 < must be provided' +[[ ${jar} == "" ]] && errors=$((errors + 1)) && echo '--jar= must be provided' +[[ ${topic_name} == "" ]] && errors=$((errors + 1)) && echo '--topic-name= 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 <"${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 From d37a973767f63ac1338099bb0ab38f7a72424ff9 Mon Sep 17 00:00:00 2001 From: Eder Ruiz Date: Fri, 4 Dec 2020 11:48:55 -0300 Subject: [PATCH 2/6] Add app config for prod env --- src/main/resources/application.conf | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/main/resources/application.conf b/src/main/resources/application.conf index 7ac9e66..a0702da 100644 --- a/src/main/resources/application.conf +++ b/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" From c853d73ade5a425e1dba84e4f98510d958a9fd62 Mon Sep 17 00:00:00 2001 From: Eder Ruiz Date: Fri, 4 Dec 2020 11:49:51 -0300 Subject: [PATCH 3/6] Fix code style --- src/main/scala/topic/reloader/main/TopicReloaderApp.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/main/scala/topic/reloader/main/TopicReloaderApp.scala b/src/main/scala/topic/reloader/main/TopicReloaderApp.scala index 8e9be76..f6d8af2 100644 --- a/src/main/scala/topic/reloader/main/TopicReloaderApp.scala +++ b/src/main/scala/topic/reloader/main/TopicReloaderApp.scala @@ -43,7 +43,7 @@ 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 reloadedTopicName = s"$topicName-reloaded" @@ -68,7 +68,7 @@ object TopicReloaderApp extends App { .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() } From b69f2a78ffeedc0eb41497eedf101d2acbd89e8e Mon Sep 17 00:00:00 2001 From: Eder Ruiz Date: Fri, 4 Dec 2020 11:52:49 -0300 Subject: [PATCH 4/6] Add support to retrieve latest schema from schema registry --- src/main/scala/topic/reloader/main/TopicReloaderApp.scala | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/main/scala/topic/reloader/main/TopicReloaderApp.scala b/src/main/scala/topic/reloader/main/TopicReloaderApp.scala index f6d8af2..3a4fe33 100644 --- a/src/main/scala/topic/reloader/main/TopicReloaderApp.scala +++ b/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 { @@ -45,8 +47,12 @@ object TopicReloaderApp extends App { "basic.auth.credentials.source" -> "USER_INFO", "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 From a5e9bd52df3cbcd92a588e47944d036763f6b333 Mon Sep 17 00:00:00 2001 From: Eder Ruiz Date: Fri, 4 Dec 2020 11:53:35 -0300 Subject: [PATCH 5/6] Add some debug info --- src/main/scala/topic/reloader/main/TopicReloaderApp.scala | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/src/main/scala/topic/reloader/main/TopicReloaderApp.scala b/src/main/scala/topic/reloader/main/TopicReloaderApp.scala index 3a4fe33..07a5fb6 100644 --- a/src/main/scala/topic/reloader/main/TopicReloaderApp.scala +++ b/src/main/scala/topic/reloader/main/TopicReloaderApp.scala @@ -66,9 +66,17 @@ object TopicReloaderApp extends App { .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) From 19c5d5ec81b54bfd6848f4dc9a17787bfd1a460e Mon Sep 17 00:00:00 2001 From: Eder Ruiz Date: Fri, 4 Dec 2020 11:55:24 -0300 Subject: [PATCH 6/6] Use latest schema version --- src/main/scala/topic/reloader/main/TopicReloaderApp.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/scala/topic/reloader/main/TopicReloaderApp.scala b/src/main/scala/topic/reloader/main/TopicReloaderApp.scala index 07a5fb6..e43f1a8 100644 --- a/src/main/scala/topic/reloader/main/TopicReloaderApp.scala +++ b/src/main/scala/topic/reloader/main/TopicReloaderApp.scala @@ -62,7 +62,7 @@ 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.