Skip to content

Slw changes #171

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
FsherinP wants to merge 4 commits into
base: cbrelease-4.8.23
Choose a base branch
from
Open

Slw changes #171

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
6b9cdf0
removing unwanted lines
FsherinP Mar 17, 2025
7091b08
maharshtra changes
FsherinP Apr 24, 2025
7445aa7
maharashtra changes
FsherinP Apr 24, 2025
8d66453
maharashtra changes
FsherinP Apr 24, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
90 changes: 75 additions & 15 deletions ...scala/org/ekstep/analytics/dashboard/nationallearningweek/NationalLearningWeekModel.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,14 +49,13 @@ object NationalLearningWeekModel extends AbsDashboardModel {
val orgHierarchyDF = cache.load("orgHierarchy")

val eventCertificatesGeneratedSLWYdayDF = eventsEnrolmentsDF
.filter(col("status") === "completed")
.filter(col("enrolled_on_datetime") >= previousStart && col("enrolled_on_datetime") <= previousEnd)
.filter(col("certificate_id").isNotNull)
.join(userDetailsDF, Seq("user_id"), "left")
.join(orgHierarchyDF, Seq("mdo_id"), "left")
.withColumn("ministry_id", coalesce(col("ministry_id"), col("mdo_id"))) // Replace null ministry_id with mdo_id
.groupBy("ministry_id")
.agg(countDistinct("certificate_id").alias("event_certificate_count"))
.filter(col("completed_on_datetime") >= previousStart && col("completed_on_datetime") <= previousEnd)
.filter(col("certificate_id").isNotNull)
.join(userDetailsDF, Seq("user_id"), "left")
.join(orgHierarchyDF, Seq("mdo_id"), "left")
.withColumn("ministry_id", coalesce(col("ministry_id"), col("mdo_id"))) // Replace null ministry_id with mdo_id
.groupBy("ministry_id")
.agg(countDistinct("certificate_id").alias("event_certificate_count"))

val contentCertificatesGeneratedSLWYdayDF = contentEnrolmentsDF
.filter(col("last_certificate_generated_on") >= previousStart && col("last_certificate_generated_on") <= previousEnd)
Expand All @@ -71,7 +70,7 @@ object NationalLearningWeekModel extends AbsDashboardModel {
val totalCertificatesGeneratedSLWYdayByOrgDF = eventCertificatesGeneratedSLWYdayDF
.join(contentCertificatesGeneratedSLWYdayDF, Seq("ministry_id"), "outer")
.withColumn("total_certificate_generatedYday_slw_count", coalesce(col("event_certificate_count"), lit(0)) +
coalesce(col("content_certificate_count"), lit(0)))
coalesce(col("content_certificate_count"), lit(0)))
.filter(col("ministry_id").isNotNull)

Redis.dispatchDataFrame[Int]("dashboard_certificate_generated_yday_by_ministry_slw_count", totalCertificatesGeneratedSLWYdayByOrgDF, "ministry_id", "total_certificate_generatedYday_slw_count")
Expand All @@ -95,10 +94,39 @@ object NationalLearningWeekModel extends AbsDashboardModel {
val totalEnrolmentsInSLWByMinistryDF = eventEnrolmentsInSLWDF
.join(contentEnrolmentsInSLWDF, Seq("ministry_id"), "full_outer")
.select(col("ministry_id"), coalesce(col("event_enrolment_count"), lit(0)).alias("event_enrolment_count"), coalesce(col("content_enrolment_count"), lit(0)).alias("content_enrolment_count"),
(coalesce(col("event_enrolment_count"), lit(0)) + coalesce(col("content_enrolment_count"), lit(0))).alias("total_enrolments"))
(coalesce(col("event_enrolment_count"), lit(0)) + coalesce(col("content_enrolment_count"), lit(0))).alias("total_enrolments"))
.filter(col("ministry_id").isNotNull)

Redis.dispatchDataFrame[Int]("dashboard_total_enrolment_by_ministry_slw_count", totalEnrolmentsInSLWByMinistryDF, "ministry_id", "total_enrolments")
// for maharashtra
val maharashtraEventEnrolments = eventsEnrolmentsDF
.filter(col("ministry_id") === "")
.filter(col("enrolled_on_datetime") >= "" && col("enrolled_on_datetime") <= "")
.join(userDetailsDF, Seq("user_id"), "left")
.join(orgHierarchyDF, Seq("mdo_id"), "left")
.withColumn("ministry_id", coalesce(col("ministry_id"), col("mdo_id")))
.groupBy("ministry_id")
.agg(count("*").alias("event_enrolment_count"))

val maharashtraContentEnrolments = contentEnrolmentsDF
.filter(col("ministry_id") === "")
.filter(col("enrolled_on") >= "" && col("enrolled_on") <= "")
.join(userDetailsDF,Seq("user_id"), "left")
.join(orgHierarchyDF, Seq("mdo_id"), "left")
.withColumn("ministry_id", coalesce(col("ministry_id"), col("mdo_id")))
.groupBy("ministry_id")
.agg(count("*").alias("content_enrolment_count"))

val maharashtraTotalEnrolments = maharashtraEventEnrolments.join(maharashtraContentEnrolments, Seq("ministry_id"), "full_outer")
.select(
col("ministry_id"),
coalesce(col("event_enrolment_count"), lit(0)).alias("event_enrolment_count"),
coalesce(col("content_enrolment_count"), lit(0)).alias("content_enrolment_count"),
(coalesce(col("event_enrolment_count"), lit(0)) + coalesce(col("content_enrolment_count"), lit(0))).alias("total_enrolments"))
.filter(col("ministry_id").isNotNull
)

Redis.dispatchDataFrame[Int]("dashboard_total_enrolment_by_ministry_slw_count", totalEnrolmentsInSLWByMinistryDF.filter(col("ministry_id") =!= ""), "ministry_id", "total_enrolments")
Redis.dispatchDataFrame[Int]("dashboard_total_enrolment_by_ministry_slw_count", maharashtraTotalEnrolments, "ministry_id", "total_enrolments")


val eventCertificatesGeneratedInSLWDF = eventsEnrolmentsDF
Expand Down Expand Up @@ -127,8 +155,38 @@ object NationalLearningWeekModel extends AbsDashboardModel {
(coalesce(col("event_certificate_count"), lit(0)) + coalesce(col("content_certificate_count"), lit(0))).alias("total_certificates"))
.filter(col("ministry_id").isNotNull)

Redis.dispatchDataFrame[Int]("dashboard_certificates_generated_by_ministry_slw_count", totalCertificatesGeneratedInSLWByMinistryDF, "ministry_id", "total_certificates")
*/
//maharashtra certificates generated
val maharshtraEventCertificates = eventsEnrolmentsDF
.filter(col("ministry_id") === "")
.filter(col("completed_on_datetime") >= "" && col("completed_on_datetime") <= "")
.filter(col("certificate_id").isNotNull)
.join(userDetailsDF, Seq("user_id"), "left")
.join(orgHierarchyDF, Seq("mdo_id"), "left")
.withColumn("ministry_id", coalesce(col("ministry_id"), col("mdo_id")))
.groupBy("ministry_id")
.agg(countDistinct("certificate_id").alias("event_certificate_count"))

val maharshtraContentCertificates = contentEnrolmentsDF
.filter(col("ministry_id") === "")
.filter(col("first_completed_on") >= "" && col("first_completed_on") <= "")
.filter(col("certificate_id").isNotNull)
.join(userDetailsDF,Seq("user_id"), "left")
.join(orgHierarchyDF, Seq("mdo_id"), "left")
.withColumn("ministry_id", coalesce(col("ministry_id"), col("mdo_id")))
.groupBy("ministry_id")
.agg(count("*").alias("content_certificate_count"))

val maharashtraTotalCertificates = maharshtraEventCertificates.join(maharshtraContentCertificates, Seq("ministry_id"), "full_outer")
.select(
col("ministry_id"),
coalesce(col("event_certificate_count"), lit(0)).alias("event_certificate_count"),
coalesce(col("content_certificate_count"), lit(0)).alias("content_certificate_count"),
(coalesce(col("event_certificate_count"), lit(0)) + coalesce(col("content_certificate_count"), lit(0))).alias("total_certificates"))
.filter(col("ministry_id").isNotNull)

Redis.dispatchDataFrame[Int]("dashboard_certificates_generated_by_ministry_slw_count", totalCertificatesGeneratedInSLWByMinistryDF.filter(col("ministry_id") =!= ""), "ministry_id", "total_certificates")
Redis.dispatchDataFrame[Int]("dashboard_certificates_generated_by_ministry_slw_count", maharashtraTotalCertificates, "ministry_id", "total_certificates")

val slwStartDate = stateLearningWeekStartString.split(" ")(0)
val slwEndDate = stateLearningWeekEndString.split(" ")(0)
val slwDateConditions = s"""{"range": {"startDate": {"gte": "${slwStartDate}", "lte": "${slwEndDate}"}}}"""
Expand All @@ -139,11 +197,13 @@ object NationalLearningWeekModel extends AbsDashboardModel {
val fieldsClauseRequired = fieldsRequired.map(f => s""""${f}"""").mkString(",")
val eventQuery = s"""{"_source":[${fieldsClauseRequired}],"query":{"bool":{"must": [${slwDateConditions}], "should":[${shouldClauseRequired}]}}}"""
val eventDataDF = elasticSearchDataFrame(conf.sparkElasticsearchConnectionHost, "compositesearch", eventQuery, fieldsRequired, arrayFieldsRequired)
val eventsPublishedDF = eventDataDF.agg(
val eventsPublishedDF = eventDataDF.filter(col("createdFor") =!= "").agg(
lit("01397282245867929648").alias("ministry_id"), count("identifier").alias("events_published_count"))
val maharshtraEventsCount = eventDataDF.filter(col("createdFor") === "").agg(
lit("01397282245867929648").alias("ministry_id"), count("identifier").alias("events_published_count"))


Redis.dispatchDataFrame[Int]("dashboard_events_published_by_ministry_slw_count", eventsPublishedDF, "ministry_id", "events_published_count")
Redis.update("dashboard_events_published_by_ministry_slw_count", maharshtraEventsCount.toString)

val userEventCertificatesDF = eventsEnrolmentsDF
.filter(col("completed_on_datetime") >= stateLearningWeekStartString && col("completed_on_datetime") <= stateLearningWeekEndString)
Expand Down