Issue #LR-491 merge: ML Program UserInfoExhaust Data Product (#40)

Author: Shakthieshwari

ansible/roles/lern-data-products-deploy/defaults/main.yml

@@ -32,6 +32,7 @@ hierarchySearchServicEndpoint: /v3/hierarchy/
 
 user_table_keyspace: "sunbird"
 course_keyspace: "sunbird_courses"
+program_keyspace: "sunbird_programs"
 hierarchy_store_keyspace: "{{ env }}_hierarchy_store"
 job_request_table: "{{ env }}_job_request"
 dataset_metadata_table: "{{ env }}_dataset_metadata"

ansible/roles/lern-data-products-deploy/templates/common.conf.j2

@@ -260,12 +260,15 @@ druid.report.default.container="report-verification"
 sunbird.user.keyspace="{{ user_table_keyspace }}"
 sunbird.courses.keyspace="{{ course_keyspace }}"
 sunbird.content.hierarchy.keyspace="{{ cassandra_hierarchy_store_keyspace }}"
+sunbird.program.report.keyspace="{{ program_keyspace }}"
 sunbird.user.cluster.host="{{ core_cassandra_host }}"
+sunbird.program.report.host="{{ core_cassandra_host }}"
 sunbird.courses.cluster.host="{{ core_cassandra_host }}"
 sunbird.content.cluster.host="{{ core_cassandra_host }}"
 sunbird.report.cluster.host="{{ report_cassandra_cluster_host }}"
 sunbird.user.report.keyspace="{{ report_user_table_keyspace }}"
 collection.exhaust.store.prefix=""
+ml.exhaust.store.prefix="ml_reports"
 postgres.table.job_request="{{ job_request_table }}"
 postgres.table.dataset_metadata="{{ dataset_metadata_table }}"
 

ansible/roles/lern-data-products-deploy/templates/lern-model-config.j2

@@ -93,6 +93,9 @@ config() {
     "old-certificate-migration-job")
     echo '{"search":{"type":"none"},"model":"org.sunbird.lms.audit.OldCertificateMigrationJob","modelParams":{"mode":"execute","store":"azure","sparkCassandraConnectionHost":"{{ core_cassandra_host }}", "cert_base_path": "https://staging.sunbirded.org/", "cloud_storage_base_url": "https://sunbirdstaging.blob.core.windows.net", "cloud_store_base_path_placeholder": "CLOUD_BASE_PATH","content_cloud_storage_container": "sunbird-content-staging", "cloud_storage_cname_url": "https://obj.dev.sunbirded.org", "batchId": "0134278454483681283", "kafka_broker": "localhost:29092"}}'
     ;;
+    "program-user-exhaust")
+    echo '{"search":{"type":"none"},"model":"org.sunbird.ml.exhaust.ProgramUserInfoExhaustJob","modelParams":{"store":"azure","mode":"OnDemand","authorizedRoles":["PROGRAM_MANAGER"],"id":"ml-program-user-exhaust","keyspace_name":"sunbird_programs","table":[{"columns":["user_id","program_name","program_externalId","user_locations","user_type","user_sub_type","organisation_name","pii_consent_required"],"name":"program_enrollment","user_locations_columns":["state_name","district_name","block_name","cluster_name","school_code","school_name"]},{"name":"user","columns":["userid","firstname","lastname","email","phone","username"],"encrypted_columns":["email","phone"],"final_columns":["email","phone","username"]}],"label_mapping":{"user_id":"User UUID","username":"User Name(On user consent)","phone":"Mobile number(On user consent)","email":"Email ID(On user consent)","consentflag":"Consent Provided","consentprovideddate":"Consent Provided Date","program_name":"Program Name","program_externalId":"Program ID","state_name":"State","district_name":"District","block_name":"Block","cluster_name":"Cluster","school_code":"School Id","school_name":"School Name","user_type":"Usertype","user_sub_type":"Usersubtype","organisation_name":"Org Name"},"order_of_csv_column":["User UUID","User Name(On user consent)","Mobile number(On user consent)","Email ID(On user consent)","Consent Provided","Consent Provided Date","Program Name","Program ID","State","District","Block","Cluster","School Id","School Name","Usertype","Usersubtype","Org Name"],"sort":["District","Block","Cluster","School Id","User UUID"],"quote_column":["User Name(On user consent)","Program Name"],"sparkElasticsearchConnectionHost":"{{ sunbird_es_host }}","sparkRedisConnectionHost":"{{ metadata2_redis_host }}","sparkUserDbRedisIndex":"12","sparkCassandraConnectionHost":"{{ core_cassandra_host }}","sparkUserDbRedisPort":"{{ user_port }}","fromDate":"$(date --date yesterday '+%Y-%m-%d')","toDate":"$(date --date yesterday '+%Y-%m-%d')","key":"ml_reports/","format":"csv"},"output":[{"params":{"file":"ml_reports/"},"to":"file"}],"parallelization":8,"appName":"Program UserInfo Exhaust"}'
+    ;;
     "*")
     echo "Unknown model code"
     exit 1 # Command to come out of the program with status 1

ansible/roles/lern-data-products-deploy/templates/lern-run-job.j2

@@ -57,6 +57,8 @@ get_report_job_model_name(){
     ;;
     "old-certificate-migration-job") echo 'org.sunbird.lms.audit.OldCertificateMigrationJob'
     ;;
+    "program-user-exhaust") echo 'org.sunbird.ml.exhaust.ProgramUserInfoExhaustJob'
+    ;;
     *) echo $1
     ;;
   esac

lern-data-products/src/main/resources/application.conf

@@ -164,10 +164,13 @@ sunbird.user.keyspace="sunbird"
 sunbird.courses.keyspace="sunbird_courses"
 sunbird.user.keyspace="sunbird"
 sunbird.content.hierarchy.keyspace="sunbird_courses"
+sunbird.program.report.keyspace="sunbird_programs"
 sunbird.user.cluster.host=127.0.0.1
 sunbird.courses.cluster.host=127.0.0.1
 sunbird.content.cluster.host=127.0.0.1
+sunbird.program.report.host=127.0.0.1
 collection.exhaust.store.prefix="reports/"
+ml.exhaust.store.prefix="ml_reports"
 postgres.table.job_request="job_request"
 
 ## Collection Exhaust Jobs Configuration - End ##

lern-data-products/src/main/resources/data.cql

@@ -536,4 +536,38 @@ CREATE TABLE IF NOT EXISTS test_keyspace.report_user_enrolments(
     progress int,
     status int,
     PRIMARY KEY (batchid, courseid, userid))
-    WITH COMPACTION = { 'class': 'SizeTieredCompactionStrategy', 'enabled': 'false' };
+    WITH COMPACTION = { 'class': 'SizeTieredCompactionStrategy', 'enabled': 'false' };
+
+CREATE KEYSPACE IF NOT EXISTS sunbird_programs WITH replication = {
+    'class': 'SimpleStrategy',
+    'replication_factor': '1'
+ };
+
+CREATE TABLE IF NOT EXISTS sunbird_programs.program_enrollment (
+    program_id text,
+    program_externalId text,
+    program_name text,
+    pii_consent_required boolean,
+    user_id text,
+    user_locations map<text,text>,
+    organisation_id text,
+    organisation_name text,
+    user_sub_type text,
+    user_type text,
+    created_at date,
+    updated_at date,
+    PRIMARY KEY (program_id,user_id)
+) WITH bloom_filter_fp_chance = 0.01
+    AND caching = {'keys': 'ALL', 'rows_per_partition': 'NONE'}
+    AND comment = ''
+    AND compaction = {'class': 'org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy', 'max_threshold': '32', 'min_threshold': '4'}
+    AND compression = {'chunk_length_in_kb': '64', 'class': 'org.apache.cassandra.io.compress.LZ4Compressor'}
+    AND crc_check_chance = 1.0
+    AND dclocal_read_repair_chance = 0.1
+    AND default_time_to_live = 0
+    AND gc_grace_seconds = 864000
+    AND max_index_interval = 2048
+    AND memtable_flush_period_in_ms = 0
+    AND min_index_interval = 128
+    AND read_repair_chance = 0.0
+    AND speculative_retry = '99PERCENTILE';

lern-data-products/src/main/scala/org/sunbird/core/exhaust/UserInfoUtil.scala

@@ -0,0 +1,45 @@
+package org.sunbird.core.exhaust
+
+import org.apache.spark.sql.{DataFrame, Encoders, SparkSession}
+import org.ekstep.analytics.framework.FrameworkContext
+import org.ekstep.analytics.framework.conf.AppConf
+import org.ekstep.analytics.framework.util.{CommonUtil, JobLogger}
+import org.apache.spark.sql._
+import org.apache.spark.sql.functions._
+import org.apache.spark.sql.types.StructType
+import org.sunbird.lms.exhaust.collection.UDFUtils
+case class UserData(userid: String, state: Option[String] = Option(""), district: Option[String] = Option(""), orgname: Option[String] = Option(""), firstname: Option[String] = Option(""), lastname: Option[String] = Option(""), email: Option[String] = Option(""),
+                    phone: Option[String] = Option(""), rootorgid: String, block: Option[String] = Option(""), schoolname: Option[String] = Option(""), schooludisecode: Option[String] = Option(""), board: Option[String] = Option(""), cluster: Option[String] = Option(""),
+                    usertype: Option[String] = Option(""), usersubtype: Option[String] = Option(""))
+object UserInfoUtil extends BaseReportsJob {
+  private val userCacheDBSettings = Map("table" -> "user", "infer.schema" -> "true", "key.column" -> "userid");
+  private val userConsentDBSettings = Map("table" -> "user_consent", "keyspace" -> AppConf.getConfig("sunbird.user.keyspace"), "cluster" -> "UserCluster");
+  private val redisFormat = "org.apache.spark.sql.redis";
+  val cassandraFormat = "org.apache.spark.sql.cassandra";
+  def getUserCacheDF(cols: Seq[String], persist: Boolean)(implicit spark: SparkSession): DataFrame = {
+    val schema = Encoders.product[UserData].schema
+    val df = loadData(userCacheDBSettings, redisFormat, schema).withColumn("username", concat_ws(" ", col("firstname"), col("lastname"))).select(cols.head, cols.tail: _*)
+      .repartition(AppConf.getConfig("exhaust.user.parallelism").toInt, col("userid"))
+    if (persist) df.persist() else df
+  }
+
+  def getUserConsentDF(filters: String,  persist: Boolean)(implicit spark: SparkSession): DataFrame = {
+    val df = loadData(userConsentDBSettings, cassandraFormat, new StructType())
+      .where(s"""$filters""")
+      .dropDuplicates("user_id", "object_id")
+      .withColumn("consentflag", when(lower(col("status")) === "active", "true").otherwise("false"))
+      .withColumn("last_updated_on", date_format(col("last_updated_on"), "dd/MM/yyyy"))
+      .select(col("user_id").as("userid"), col("consentflag"), col("last_updated_on").as("consentprovideddate"));
+    if (persist) df.persist() else df
+  }
+
+  def decryptUserInfo(pgmUserDF: DataFrame, userCacheEncryptColNames: List[String])(implicit spark: SparkSession): DataFrame = {
+    val schema = pgmUserDF.schema
+    val decryptFields = schema.fields.filter(field => userCacheEncryptColNames.contains(field.name));
+    var resultDF = decryptFields.foldLeft(pgmUserDF)((df, field) => {
+      df.withColumn(field.name, when(col("consentflag") === "true", UDFUtils.toDecrypt(col(field.name))).otherwise(lit("")))
+    })
+    resultDF = resultDF.withColumn("username",when(col("consentflag") === "true", col("username")).otherwise(lit("")))
+    resultDF
+  }
+}