Automating Spark Jobs with GitLab CI/CD

من إيلوم . واجهة برمجة التطبيقات استورد Ilum Job 
من بايسبارك . SQL . functions استورد  col,  sum مثل  spark_sum
من  io استورد  StringIO

فصل  SparkInteractiveExample( Ilum Job ) : 
    مواطنه  ركض ( ذات , شراره , التكوين )  - >  str: 
table_name = التكوين . حصل ( "الجدول" ) 
        database_name = التكوين . حصل ( 'database')   # optional
        report_lines =  [ ] 

        لو  لا table_name : 
            raise ValueError( "Config must provide a 'table' key") 

        # Use specified database if provided
        لو  database_name: 
شراره . كتالوج . setCurrentDatabase( database_name) 
            report_lines. append( f"Using database: { database_name} " ) 

        # Check if table exists in catalog
        لو table_name لا  في  [ t . اسم من أجل t في شراره . كتالوج . listTables( ) ] : 
            raise ValueError( f"Table '{ table_name } ' not found in catalog") 

مدافع = شراره . جدول ( table_name ) 

        report_lines. append( f"=== Details for table: { table_name }  ===") 

        # Total rows
        total_rows = مدافع . عد ( ) 
        report_lines. append( f"Total rows: { total_rows} " ) 

        # Total columns
        total_columns =  len( مدافع . columns) 
        report_lines. append( f"Total columns: { total_columns} " ) 

        # Distinct counts per column
        report_lines. append( "Distinct values per column:") 
        من أجل c في مدافع . columns: 
            distinct_count = مدافع . select( c ) . distinct( ) . عد ( ) 
            report_lines. append( f"  { c } : { distinct_count} " ) 

        # Schema info
        report_lines. append( "Schema:") 
        # Spark does not easily return schema as string; we can reconstruct:
        من أجل f في مدافع . مخطط . fields: 
            report_lines. append( f"  { f . اسم } : { f . dataType} " ) 

        # Sample data
        report_lines. append( "Sample data (first 5 rows):") 
        sample_rows = مدافع . take( 5 ) 
        من أجل صف في  sample_rows: 
            report_lines. append( str( صف . asDict( ) ) ) 

        # Null counts per column
        report_lines. append( "Null counts per column:") 
        null_counts_df = مدافع . select( [ spark_sum( col( c ) . isNull( ) . cast( "int") ) . alias( c )  من أجل c في مدافع . columns] ) 
        null_counts =  null_counts_df. collect( ) [ 0 ] . asDict( ) 
        من أجل c ,  v في  null_counts. items( ) : 
            report_lines. append( f"  { c } : { v} " ) 

        أعاد  "\n". join( report_lines) 

استورد  logging
من بايسبارك . SQL استورد جلسة سبارك 

logging. basicConfig( level= logging. INFO) 
logger =  logging. getLogger( __اسم__ ) 

logger. معلومات ( """
=== Example Spark Job: Student Enrollments ===
This Spark job demonstrates a simple educational data pipeline using Hive tables.
It performs the following steps:
1. Creates a 'students' table with student information.
2. Creates a 'courses' table with available courses.
3. Creates an 'enrollments' table linking students to courses.
4. Joins the tables to calculate enrollment statistics and saves them into 'course_stats'.

All intermediate results are stored in the 'university' Hive database.
=============================================
""") 

شراره =  SparkSession \
    . بان\ 
    . appName ( "Spark") \ 
    . getOrCreate ( ) 

logger. معلومات ( "SparkSession initialized") 

شراره . SQL ( "CREATE DATABASE IF NOT EXISTS university") 
logger. معلومات ( "Database 'university' ensured") 

# --- Create Students Table ---
students_data =  [ 
    ( 1 ,  "Alice",  "Computer Science") , 
    ( 2 ,  "Bob",  "Mathematics") , 
    ( 3 ,  "Charlie",  "Physics") , 
    ( 4 ,  "Diana",  "Computer Science") 
] 

df_students = شراره . createDataFrame ( students_data,  [ "student_id",  "الاسم" ,  "major"] ) 
شراره . SQL ( "DROP TABLE IF EXISTS university.students") 
df_students. يكتب . format( "delta") . saveAsTable( "university.students") 
logger. معلومات ( "Created table: university.students") 

# --- Create Courses Table ---
courses_data =  [ 
    ( 101,  "Big Data") , 
    ( 102,  "Linear Algebra") , 
    ( 103,  "Quantum Mechanics") 
] 

df_courses = شراره . createDataFrame ( courses_data,  [ "course_id",  "course_name"] ) 
شراره . SQL ( "DROP TABLE IF EXISTS university.courses") 
df_courses. يكتب . format( "delta") . saveAsTable( "university.courses") 
logger. معلومات ( "Created table: university.courses") 

# --- Create Enrollments Table ---
enrollments_data =  [ 
    ( 1 ,  101) ,   # Alice -> Big Data
    ( 2 ,  102) ,   # Bob -> Linear Algebra
    ( 3 ,  103) ,   # Charlie -> Quantum Mechanics
    ( 4 ,  101) ,   # Diana -> Big Data
    ( 2 ,  101)    # Bob -> Big Data
] 

df_enrollments = شراره . createDataFrame ( enrollments_data,  [ "student_id",  "course_id"] ) 
شراره . SQL ( "DROP TABLE IF EXISTS university.enrollments") 
df_enrollments. يكتب . format( "delta") . saveAsTable( "university.enrollments") 
logger. معلومات ( "Created table: university.enrollments") 

# --- Join to calculate course enrollment counts ---
df_course_stats = شراره . SQL ( """
اختار 
  c.course_id,
  c.course_name,
  COUNT(e.student_id) AS total_students
FROM university.courses c
LEFT JOIN university.enrollments e ON c.course_id = e.course_id
GROUP BY c.course_id, c.course_name
""") 

شراره . SQL ( "DROP TABLE IF EXISTS university.course_stats") 
df_course_stats. يكتب . format( "delta") . saveAsTable( "university.course_stats") 
logger. معلومات ( "Inserted final data into course_stats table")