Building NYC Taxi Data Pipeline with Spark and Kafka

 1Data Sources
 2├── NYC TLC Trip Records (Batch)
 3├── Real-time Taxi Events (Stream, simulated: TLC publishes no real-time feed)
 4├── Weather Data (External API)
 5└── Traffic Patterns (IoT Sensors)
 6
 7Batch Processing Layer
 8├── Apache Spark 3.1 (ships with EMR 6.3.0, used below)
 9├── Hadoop HDFS 3.x
10├── Apache Hive 3.x
11├── Python 3.8 / Scala 2.12
12└── AWS EMR Clusters
13
14Stream Processing Layer
15├── Apache Kafka 2.8
16├── Spark Structured Streaming 3.1
17├── Apache Zookeeper 3.7
18├── AWS Kinesis Data Streams
19└── Flink (Alternative Processing)
20
21Storage Layer
22├── Amazon S3 (Data Lake)
23├── MySQL 8.0 (Transactional)
24├── Apache Hive (Analytics)
25├── AWS DynamoDB (NoSQL)
26└── Elasticsearch 7.x (Search)
27
28Monitoring & Analytics
29├── ELK Stack (Elasticsearch, Logstash, Kibana)
30├── Prometheus (Metrics Collection)
31├── Grafana (Visualization)
32├── AWS CloudWatch (Infrastructure)
33└── Custom Dashboards
34
35Infrastructure
36├── AWS EMR (Managed Hadoop)
37├── EC2 Instances (Compute)
38├── Auto Scaling Groups
39├── VPC Networking
40└── IAM Security

  1import org.apache.spark.sql.{SparkSession, DataFrame}
  2import org.apache.spark.sql.functions._
  3import org.apache.spark.sql.types._
  4import java.time.LocalDateTime
  5
  6object TaxiDataProcessor {
  7
  8  case class TripRecord(
  9    vendorId: Int,
 10    pickupDateTime: String,
 11    dropoffDateTime: String,
 12    passengerCount: Int,
 13    tripDistance: Double,
 14    pickupLongitude: Double,
 15    pickupLatitude: Double,
 16    dropoffLongitude: Double,
 17    dropoffLatitude: Double,
 18    paymentType: Int,
 19    fareAmount: Double,
 20    extra: Double,
 21    mtaTax: Double,
 22    tipAmount: Double,
 23    tollsAmount: Double,
 24    totalAmount: Double
 25  )
 26
 27  def main(args: Array[String]): Unit = {
 28    val spark = createSparkSession()
 29
 30    try {
 31      // Process daily batch
 32      val inputPath = args(0)  // S3 path to raw data
 33      val outputPath = args(1) // S3 path for processed data
 34
 35      val tripData = loadTripData(spark, inputPath)
 36      val cleanedData = cleanAndValidateData(tripData)
 37      val aggregatedData = performAggregations(cleanedData)
 38
 39      // Save to multiple destinations
 40      saveToHive(spark, aggregatedData)
 41      saveToMySQL(aggregatedData)
 42      saveToS3(aggregatedData, outputPath)
 43
 44      // Generate analytics reports
 45      generateDailyReport(spark, cleanedData)
 46      updateDriverRankings(spark, cleanedData)
 47
 48      logger.info("Batch processing completed successfully")
 49
 50    } catch {
 51      case e: Exception =>
 52        logger.error(s"Batch processing failed: ${e.getMessage}", e)
 53        throw e
 54    } finally {
 55      spark.stop()
 56    }
 57  }
 58
 59  /**
 60   * Create optimized Spark session for batch processing
 61   */
 62  def createSparkSession(): SparkSession = {
 63    SparkSession.builder()
 64      .appName("NYC-Taxi-Batch-Pipeline")
 65      .config("spark.sql.adaptive.enabled", "true")
 66      .config("spark.sql.adaptive.coalescePartitions.enabled", "true")
 67      .config("spark.sql.adaptive.skewJoin.enabled", "true")
 68      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
 69      .config("spark.sql.execution.arrow.pyspark.enabled", "true")
 70      .config("spark.dynamicAllocation.enabled", "true")
 71      .config("spark.dynamicAllocation.minExecutors", "2")
 72      .config("spark.dynamicAllocation.maxExecutors", "50")
 73      .config("spark.dynamicAllocation.initialExecutors", "10")
 74      .getOrCreate()
 75  }
 76
 77  /**
 78   * Load and parse taxi trip data from S3
 79   */
 80  def loadTripData(spark: SparkSession, inputPath: String): DataFrame = {
 81    spark.read
 82      .option("header", "true")
 83      .option("inferSchema", "true")
 84      .option("timestampFormat", "yyyy-MM-dd HH:mm:ss")
 85      .csv(inputPath)
 86      .filter(col("pickup_datetime").isNotNull)
 87      .filter(col("dropoff_datetime").isNotNull)
 88      .repartition(200, col("pickup_datetime"))  // Optimize partitioning
 89  }
 90
 91  /**
 92   * Data cleaning and validation pipeline
 93   */
 94  def cleanAndValidateData(df: DataFrame): DataFrame = {
 95    df
 96      .filter(col("trip_distance") > 0 && col("trip_distance") < 100)  // Reasonable distance
 97      .filter(col("fare_amount") > 0 && col("fare_amount") < 500)      // Reasonable fare
 98      .filter(col("passenger_count") > 0 && col("passenger_count") <= 9) // Valid passenger count
 99      .filter(col("pickup_latitude").between(40.5, 40.9))             // NYC bounds
100      .filter(col("pickup_longitude").between(-74.3, -73.7))
101      .filter(col("dropoff_latitude").between(40.5, 40.9))
102      .filter(col("dropoff_longitude").between(-74.3, -73.7))
103      .withColumn("trip_duration",
104        (unix_timestamp(col("dropoff_datetime")) -
105         unix_timestamp(col("pickup_datetime"))) / 60)  // Duration in minutes
106      .filter(col("trip_duration") > 1 && col("trip_duration") < 480)  // 1 min to 8 hours
107      .withColumn("hour_of_day", hour(col("pickup_datetime")))
108      .withColumn("day_of_week", dayofweek(col("pickup_datetime")))
109      .withColumn("pickup_zone", getPickupZone(col("pickup_latitude"), col("pickup_longitude")))
110      .withColumn("dropoff_zone", getDropoffZone(col("dropoff_latitude"), col("dropoff_longitude")))
111  }
112
113  /**
114   * Comprehensive data aggregations
115   */
116  def performAggregations(df: DataFrame): DataFrame = {
117    val dailyAggregations = df
118      .withColumn("date", to_date(col("pickup_datetime")))
119      .groupBy("date", "pickup_zone")
120      .agg(
121        count("*").alias("total_trips"),
122        sum("fare_amount").alias("total_revenue"),
123        avg("fare_amount").alias("avg_fare"),
124        avg("trip_distance").alias("avg_distance"),
125        avg("trip_duration").alias("avg_duration"),
126        sum("tip_amount").alias("total_tips"),
127        countDistinct("vendor_id").alias("active_vendors")
128      )
129      .orderBy("date", "pickup_zone")
130
131    // Hourly patterns
132    val hourlyPatterns = df
133      .withColumn("date", to_date(col("pickup_datetime")))
134      .groupBy("date", "hour_of_day", "pickup_zone")
135      .agg(
136        count("*").alias("hourly_trips"),
137        avg("fare_amount").alias("hourly_avg_fare")
138      )
139
140    // Driver performance (using vendor_id as proxy)
141    val driverMetrics = df
142      .groupBy("vendor_id", to_date(col("pickup_datetime")).alias("date"))
143      .agg(
144        count("*").alias("trips_completed"),
145        sum("fare_amount").alias("daily_earnings"),
146        avg("trip_duration").alias("avg_trip_time"),
147        (sum("tip_amount") / sum("fare_amount") * 100).alias("tip_percentage")
148      )
149      .withColumn("efficiency_score",
150        col("trips_completed") * 0.4 +
151        col("daily_earnings") * 0.3 +
152        (1 / col("avg_trip_time")) * 0.3)
153
154    dailyAggregations
155  }
156
157  /**
158   * Save processed data to Hive for analytics
159   */
160  def saveToHive(spark: SparkSession, df: DataFrame): Unit = {
161    df.write
162      .mode("append")
163      .partitionBy("date")
164      .option("compression", "snappy")
165      .saveAsTable("taxi_analytics.daily_trip_summary")
166  }
167
168  /**
169   * Save aggregated metrics to MySQL for applications
170   */
171  def saveToMySQL(df: DataFrame): Unit = {
172    val connectionProperties = new java.util.Properties()
173    connectionProperties.put("user", sys.env("MYSQL_USER"))
174    connectionProperties.put("password", sys.env("MYSQL_PASSWORD"))
175    connectionProperties.put("driver", "com.mysql.cj.jdbc.Driver")
176    connectionProperties.put("rewriteBatchedStatements", "true")
177    connectionProperties.put("useSSL", "false")
178
179    df.write
180      .mode("append")
181      .option("batchsize", "10000")
182      .option("truncate", "true")
183      .jdbc(
184        url = sys.env("MYSQL_URL"),
185        table = "trip_analytics",
186        connectionProperties = connectionProperties
187      )
188  }
189
190  /**
191   * Save to S3 for data lake storage
192   */
193  def saveToS3(df: DataFrame, outputPath: String): Unit = {
194    df.write
195      .mode("overwrite")
196      .partitionBy("date", "pickup_zone")
197      .option("compression", "gzip")
198      .parquet(outputPath)
199  }
200
201  /**
202   * Generate comprehensive daily reports
203   */
204  def generateDailyReport(spark: SparkSession, df: DataFrame): Unit = {
205    import spark.implicits._
206
207    // Top performing zones
208    val topZones = df
209      .groupBy("pickup_zone")
210      .agg(
211        count("*").alias("total_trips"),
212        sum("fare_amount").alias("total_revenue")
213      )
214      .orderBy(desc("total_revenue"))
215      .limit(10)
216
217    // Peak hour analysis
218    val peakHours = df
219      .groupBy("hour_of_day")
220      .agg(
221        count("*").alias("trips"),
222        avg("fare_amount").alias("avg_fare")
223      )
224      .orderBy(desc("trips"))
225
226    // Revenue trends
227    val revenueByHour = df
228      .withColumn("date_hour", date_format(col("pickup_datetime"), "yyyy-MM-dd HH"))
229      .groupBy("date_hour")
230      .agg(
231        sum("fare_amount").alias("hourly_revenue"),
232        count("*").alias("hourly_trips")
233      )
234      .orderBy("date_hour")
235
236    // Save reports
237    topZones.coalesce(1).write.mode("overwrite").csv("s3://taxi-reports/top-zones/")
238    peakHours.coalesce(1).write.mode("overwrite").csv("s3://taxi-reports/peak-hours/")
239    revenueByHour.write.mode("overwrite").partitionBy("date_hour").parquet("s3://taxi-reports/revenue-trends/")
240  }
241
242  /**
243   * Calculate pickup zone based on coordinates
244   */
245  def getPickupZone = udf((lat: Double, lon: Double) => {
246    (lat, lon) match {
247      case (l, o) if l >= 40.75 && l <= 40.80 && o >= -74.0 && o <= -73.95 => "Manhattan_Midtown"
248      case (l, o) if l >= 40.70 && l <= 40.75 && o >= -74.0 && o <= -73.95 => "Manhattan_Downtown"
249      case (l, o) if l >= 40.80 && l <= 40.85 && o >= -74.0 && o <= -73.95 => "Manhattan_Uptown"
250      case (l, o) if l >= 40.65 && l <= 40.72 && o >= -74.0 && o <= -73.80 => "Brooklyn"
251      case (l, o) if l >= 40.72 && l <= 40.80 && o >= -73.95 && o <= -73.75 => "Queens"
252      case (l, o) if l >= 40.80 && l <= 40.90 && o >= -73.90 && o <= -73.80 => "Bronx"
253      case _ => "Other"
254    }
255  })
256
257  /**
258   * Calculate dropoff zone based on coordinates
259   */
260  def getDropoffZone = udf((lat: Double, lon: Double) => {
261    // Same logic as pickup zone
262    getPickupZone.apply(lat, lon)
263  })
264}
265
266/**
267 * Configuration and utilities
268 */
269object TaxiDataConfig {
270
271  case class ProcessingConfig(
272    inputPath: String,
273    outputPath: String,
274    checkpointLocation: String,
275    batchInterval: String,
276    maxRecordsPerTrigger: Long
277  )
278
279  def loadConfig(): ProcessingConfig = {
280    ProcessingConfig(
281      inputPath = sys.env.getOrElse("INPUT_PATH", "s3a://nyc-tlc/trip-data/"),
282      outputPath = sys.env.getOrElse("OUTPUT_PATH", "s3a://processed-taxi-data/"),
283      checkpointLocation = sys.env.getOrElse("CHECKPOINT_PATH", "s3a://taxi-checkpoints/"),
284      batchInterval = sys.env.getOrElse("BATCH_INTERVAL", "10 minutes"),
285      maxRecordsPerTrigger = sys.env.getOrElse("MAX_RECORDS_PER_TRIGGER", "100000").toLong
286    )
287  }
288}

  1import org.apache.spark.sql.{SparkSession, DataFrame}
  2import org.apache.spark.sql.streaming.{StreamingQuery, Trigger}
  3import org.apache.spark.sql.functions._
  4import org.apache.spark.sql.types._
  5import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
  6import java.util.Properties
  7
  8object TaxiStreamProcessor {
  9
 10  case class TaxiEvent(
 11    tripId: String,
 12    vendorId: Int,
 13    eventType: String,  // "pickup", "dropoff", "location_update"
 14    timestamp: Long,
 15    latitude: Double,
 16    longitude: Double,
 17    passengerCount: Int,
 18    fareAmount: Option[Double] = None,
 19    paymentType: Option[String] = None
 20  )
 21
 22  case class TripMetrics(
 23    zone: String,
 24    hour: Int,
 25    totalTrips: Long,
 26    avgFare: Double,
 27    avgDuration: Double,
 28    lastUpdate: Long
 29  )
 30
 31  def main(args: Array[String]): Unit = {
 32    val spark = createSparkSession()
 33    val config = TaxiDataConfig.loadConfig()
 34
 35    try {
 36      // Start real-time streaming pipelines
 37      val tripEventStream = processTripEvents(spark, config)
 38      val analyticsStream = processRealTimeAnalytics(spark, config)
 39      val anomalyStream = processAnomalyDetection(spark, config)
 40
 41      // Start all streaming queries
 42      val queries = List(tripEventStream, analyticsStream, anomalyStream)
 43
 44      // Wait for termination
 45      queries.foreach(_.awaitTermination())
 46
 47    } catch {
 48      case e: Exception =>
 49        logger.error(s"Stream processing failed: ${e.getMessage}", e)
 50        throw e
 51    } finally {
 52      spark.stop()
 53    }
 54  }
 55
 56  /**
 57   * Create optimized Spark session for streaming
 58   */
 59  def createSparkSession(): SparkSession = {
 60    SparkSession.builder()
 61      .appName("NYC-Taxi-Stream-Pipeline")
 62      .config("spark.sql.streaming.checkpointLocation", "/tmp/checkpoints")
 63      .config("spark.sql.streaming.forceDeleteTempCheckpointLocation", "true")
 64      .config("spark.sql.adaptive.enabled", "true")
 65      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
 66      .config("spark.sql.streaming.stateStore.maintenanceInterval", "600s")
 67      .config("spark.sql.streaming.statefulOperator.checkCorrectness.enabled", "false")
 68      .getOrCreate()
 69  }
 70
 71  /**
 72   * Process individual trip events from Kafka
 73   */
 74  def processTripEvents(spark: SparkSession, config: ProcessingConfig): StreamingQuery = {
 75    import spark.implicits._
 76
 77    // Define schema for taxi events
 78    val taxiEventSchema = StructType(Array(
 79      StructField("trip_id", StringType, true),
 80      StructField("vendor_id", IntegerType, true),
 81      StructField("event_type", StringType, true),
 82      StructField("timestamp", LongType, true),
 83      StructField("latitude", DoubleType, true),
 84      StructField("longitude", DoubleType, true),
 85      StructField("passenger_count", IntegerType, true),
 86      StructField("fare_amount", DoubleType, true),
 87      StructField("payment_type", StringType, true)
 88    ))
 89
 90    // Read from Kafka topic
 91    val kafkaStream = spark
 92      .readStream
 93      .format("kafka")
 94      .option("kafka.bootstrap.servers", sys.env("KAFKA_BROKERS"))
 95      .option("subscribe", "taxi-events")
 96      .option("startingOffsets", "latest")
 97      .option("maxOffsetsPerTrigger", config.maxRecordsPerTrigger)
 98      .load()
 99
100    // Parse and process events
101    val parsedStream = kafkaStream
102      .select(from_json(col("value").cast("string"), taxiEventSchema).alias("data"))
103      .select("data.*")
104      .withColumn("processing_time", current_timestamp())
105      .withColumn("zone", getZoneFromCoordinates($"latitude", $"longitude"))
106      .withColumn("hour_of_day", hour(from_unixtime($"timestamp" / 1000)))
107
108    // Enrich with real-time data
109    val enrichedStream = parsedStream
110      .withColumn("weather_condition", getWeatherCondition($"latitude", $"longitude", $"timestamp"))
111      .withColumn("traffic_level", getTrafficLevel($"zone", $"hour_of_day"))
112      .filter($"latitude".between(40.5, 40.9) && $"longitude".between(-74.3, -73.7))
113
114    // Write to multiple sinks
115    enrichedStream.writeStream
116      .outputMode("append")
117      .format("console")  // For debugging
118      .trigger(Trigger.ProcessingTime("10 seconds"))
119      .start()
120
121    // Save to DynamoDB for real-time lookups
122    enrichedStream.writeStream
123      .outputMode("append")
124      .foreachBatch { (batchDF: DataFrame, batchId: Long) =>
125        saveToDynamoDB(batchDF, "taxi_events_realtime")
126      }
127      .trigger(Trigger.ProcessingTime("30 seconds"))
128      .start()
129
130    // Save to Elasticsearch for search and analytics
131    enrichedStream.writeStream
132      .outputMode("append")
133      .foreachBatch { (batchDF: DataFrame, batchId: Long) =>
134        saveToElasticsearch(batchDF, "taxi-events-live")
135      }
136      .trigger(Trigger.ProcessingTime("60 seconds"))
137      .start()
138  }
139
140  /**
141   * Process real-time analytics and aggregations
142   */
143  def processRealTimeAnalytics(spark: SparkSession, config: ProcessingConfig): StreamingQuery = {
144    import spark.implicits._
145
146    val kafkaStream = spark
147      .readStream
148      .format("kafka")
149      .option("kafka.bootstrap.servers", sys.env("KAFKA_BROKERS"))
150      .option("subscribe", "taxi-events")
151      .option("startingOffsets", "latest")
152      .load()
153
154    val taxiEvents = kafkaStream
155      .select(from_json(col("value").cast("string"), getTaxiEventSchema()).alias("data"))
156      .select("data.*")
157      .withColumn("zone", getZoneFromCoordinates($"latitude", $"longitude"))
158      .withColumn("event_time", to_timestamp(from_unixtime($"timestamp" / 1000)))
159      .withWatermark("event_time", "5 minutes")
160
161    // Real-time trip counts by zone (5-minute windows)
162    val tripCounts = taxiEvents
163      .filter($"event_type" === "pickup")
164      .groupBy(
165        window($"event_time", "5 minutes", "1 minute"),
166        $"zone"
167      )
168      .agg(
169        count("*").alias("trip_count"),
170        avg("passenger_count").alias("avg_passengers"),
171        approx_count_distinct("vendor_id").alias("active_drivers")
172      )
173      .select(
174        $"window.start".alias("window_start"),
175        $"window.end".alias("window_end"),
176        $"zone",
177        $"trip_count",
178        $"avg_passengers",
179        $"active_drivers"
180      )
181
182    // Real-time revenue tracking (10-minute windows)
183    val revenueMetrics = taxiEvents
184      .filter($"event_type" === "dropoff" && $"fare_amount".isNotNull)
185      .groupBy(
186        window($"event_time", "10 minutes", "2 minutes"),
187        $"zone"
188      )
189      .agg(
190        sum("fare_amount").alias("total_revenue"),
191        avg("fare_amount").alias("avg_fare"),
192        count("*").alias("completed_trips"),
193        stddev("fare_amount").alias("fare_stddev")
194      )
195
196    // Demand forecasting (15-minute windows)
197    val demandForecast = taxiEvents
198      .filter($"event_type" === "pickup")
199      .withColumn("hour_of_day", hour($"event_time"))
200      .withColumn("day_of_week", dayofweek($"event_time"))
201      .groupBy(
202        window($"event_time", "15 minutes"),
203        $"zone",
204        $"hour_of_day",
205        $"day_of_week"
206      )
207      .agg(
208        count("*").alias("current_demand"),
209        lag(count("*"), 1).over(
210          Window.partitionBy($"zone", $"hour_of_day", $"day_of_week")
211                .orderBy($"window.start")
212        ).alias("previous_demand")
213      )
214      .withColumn("demand_trend",
215        ($"current_demand" - coalesce($"previous_demand", lit(0))) /
216        coalesce($"previous_demand", lit(1))
217      )
218
219    // Write analytics to Kafka for downstream consumption
220    tripCounts.selectExpr("CAST(zone AS STRING) AS key", "to_json(struct(*)) AS value")
221      .writeStream
222      .format("kafka")
223      .option("kafka.bootstrap.servers", sys.env("KAFKA_BROKERS"))
224      .option("topic", "trip-counts-realtime")
225      .trigger(Trigger.ProcessingTime("30 seconds"))
226      .start()
227
228    // Write to Redis for real-time dashboards
229    revenueMetrics.writeStream
230      .outputMode("update")
231      .foreachBatch { (batchDF: DataFrame, batchId: Long) =>
232        saveToRedis(batchDF, "revenue_metrics")
233      }
234      .trigger(Trigger.ProcessingTime("60 seconds"))
235      .start()
236  }
237
238  /**
239   * Real-time anomaly detection
240   */
241  def processAnomalyDetection(spark: SparkSession, config: ProcessingConfig): StreamingQuery = {
242    import spark.implicits._
243
244    val kafkaStream = spark
245      .readStream
246      .format("kafka")
247      .option("kafka.bootstrap.servers", sys.env("KAFKA_BROKERS"))
248      .option("subscribe", "taxi-events")
249      .option("startingOffsets", "latest")
250      .load()
251
252    val taxiEvents = kafkaStream
253      .select(from_json(col("value").cast("string"), getTaxiEventSchema()).alias("data"))
254      .select("data.*")
255      .withColumn("event_time", to_timestamp(from_unixtime($"timestamp" / 1000)))
256      .withWatermark("event_time", "2 minutes")
257
258    // Detect unusual fare amounts (statistical outliers)
259    val fareAnomalies = taxiEvents
260      .filter($"event_type" === "dropoff" && $"fare_amount".isNotNull)
261      .withColumn("z_score",
262        (col("fare_amount") - mean("fare_amount").over(Window.partitionBy("zone"))) /
263        stddev("fare_amount").over(Window.partitionBy("zone"))
264      )
265      .filter(abs($"z_score") > 3.0)  // Outliers beyond 3 standard deviations
266      .select(
267        $"trip_id",
268        $"vendor_id",
269        $"zone",
270        $"fare_amount",
271        $"z_score",
272        $"event_time",
273        lit("FARE_ANOMALY").alias("anomaly_type")
274      )
275
276    // Detect suspicious trip patterns (too fast/slow)
277    val speedAnomalies = taxiEvents
278      .filter($"event_type" === "dropoff")
279      .withColumn("trip_distance_km", $"trip_distance" * 1.609344)  // Miles to km
280      .withColumn("trip_duration_hours", $"trip_duration" / 60.0)    // Minutes to hours
281      .withColumn("avg_speed_kmh", $"trip_distance_km" / $"trip_duration_hours")
282      .filter($"avg_speed_kmh" > 100 || $"avg_speed_kmh" < 2)  // Unrealistic speeds
283      .select(
284        $"trip_id",
285        $"vendor_id",
286        $"avg_speed_kmh",
287        $"trip_distance_km",
288        $"trip_duration_hours",
289        $"event_time",
290        lit("SPEED_ANOMALY").alias("anomaly_type")
291      )
292
293    // Combine all anomalies
294    val allAnomalies = fareAnomalies.unionByName(speedAnomalies)
295
296    // Send alerts to monitoring system
297    allAnomalies.writeStream
298      .outputMode("append")
299      .foreachBatch { (batchDF: DataFrame, batchId: Long) =>
300        sendAnomalyAlerts(batchDF)
301        saveAnomaliesForInvestigation(batchDF)
302      }
303      .trigger(Trigger.ProcessingTime("30 seconds"))
304      .start()
305  }
306
307  /**
308   * Save processed data to DynamoDB for real-time access
309   */
310  def saveToDynamoDB(df: DataFrame, tableName: String): Unit = {
311    df.foreachPartition { partition =>
312      val dynamoClient = createDynamoDBClient()
313
314      partition.foreach { row =>
315        val item = Map(
316          "trip_id" -> AttributeValue.builder().s(row.getAs[String]("trip_id")).build(),
317          "timestamp" -> AttributeValue.builder().n(row.getAs[Long]("timestamp").toString).build(),
318          "zone" -> AttributeValue.builder().s(row.getAs[String]("zone")).build(),
319          "event_type" -> AttributeValue.builder().s(row.getAs[String]("event_type")).build(),
320          "latitude" -> AttributeValue.builder().n(row.getAs[Double]("latitude").toString).build(),
321          "longitude" -> AttributeValue.builder().n(row.getAs[Double]("longitude").toString).build()
322        )
323
324        val putRequest = PutItemRequest.builder()
325          .tableName(tableName)
326          .item(item.asJava)
327          .build()
328
329        try {
330          dynamoClient.putItem(putRequest)
331        } catch {
332          case e: Exception =>
333            logger.error(s"Failed to save item to DynamoDB: ${e.getMessage}")
334        }
335      }
336    }
337  }
338
339  /**
340   * Save data to Elasticsearch for search and visualization
341   */
342  def saveToElasticsearch(df: DataFrame, indexName: String): Unit = {
343    df.write
344      .format("org.elasticsearch.spark.sql")
345      .option("es.resource", s"$indexName/_doc")
346      .option("es.nodes", sys.env("ELASTICSEARCH_NODES"))
347      .option("es.port", sys.env.getOrElse("ELASTICSEARCH_PORT", "9200"))
348      .option("es.index.auto.create", "true")
349      .option("es.write.operation", "create")
350      .mode("append")
351      .save()
352  }
353
354  /**
355   * Send anomaly alerts to monitoring system
356   */
357  def sendAnomalyAlerts(anomalies: DataFrame): Unit = {
358    anomalies.collect().foreach { anomaly =>
359      val alert = AnomalyAlert(
360        tripId = anomaly.getAs[String]("trip_id"),
361        anomalyType = anomaly.getAs[String]("anomaly_type"),
362        severity = "HIGH",
363        timestamp = anomaly.getAs[java.sql.Timestamp]("event_time"),
364        details = anomaly.toJson
365      )
366
367      // Send to alerting system (Kafka, SNS, etc.)
368      publishAlert(alert)
369    }
370  }
371
372  /**
373   * Get zone from coordinates using UDF
374   */
375  def getZoneFromCoordinates = udf((lat: Double, lon: Double) => {
376    // Implementation similar to batch processing
377    determineZone(lat, lon)
378  })
379
380  /**
381   * Helper functions for data enrichment
382   */
383  def getWeatherCondition = udf((lat: Double, lon: Double, timestamp: Long) => {
384    // Call external weather API or use cached weather data
385    "clear"  // Simplified for example
386  })
387
388  def getTrafficLevel = udf((zone: String, hour: Int) => {
389    // Determine traffic level based on historical patterns
390    if (hour >= 7 && hour <= 10 || hour >= 17 && hour <= 20) "high" else "normal"
391  })
392}
393
394/**
395 * Real-time event producer for testing
396 */
397object TaxiEventProducer {
398
399  def main(args: Array[String]): Unit = {
400    val producer = createKafkaProducer()
401    val eventGenerator = new TaxiEventGenerator()
402
403    try {
404      while (true) {
405        val event = eventGenerator.generateRandomEvent()
406        val record = new ProducerRecord[String, String](
407          "taxi-events",
408          event.tripId,
409          event.toJson
410        )
411
412        producer.send(record)
413        Thread.sleep(100)  // Generate event every 100ms
414      }
415    } finally {
416      producer.close()
417    }
418  }
419
420  def createKafkaProducer(): KafkaProducer[String, String] = {
421    val props = new Properties()
422    props.put("bootstrap.servers", sys.env("KAFKA_BROKERS"))
423    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
424    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
425    props.put("acks", "all")
426    props.put("retries", "3")
427    props.put("batch.size", "16384")
428    props.put("linger.ms", "1")
429
430    new KafkaProducer[String, String](props)
431  }
432}
433
434/**
435 * Random taxi event generator for testing
436 */
437class TaxiEventGenerator {
438  private val random = new scala.util.Random()
439  private val nycBounds = (40.5, 40.9, -74.3, -73.7)  // lat_min, lat_max, lon_min, lon_max
440  private val zones = List("Manhattan_Midtown", "Manhattan_Downtown", "Brooklyn", "Queens", "Bronx")
441
442  def generateRandomEvent(): TaxiEvent = {
443    TaxiEvent(
444      tripId = java.util.UUID.randomUUID().toString,
445      vendorId = random.nextInt(3) + 1,
446      eventType = if (random.nextBoolean()) "pickup" else "dropoff",
447      timestamp = System.currentTimeMillis(),
448      latitude = nycBounds._1 + random.nextDouble() * (nycBounds._2 - nycBounds._1),
449      longitude = nycBounds._3 + random.nextDouble() * (nycBounds._4 - nycBounds._3),
450      passengerCount = random.nextInt(6) + 1,
451      fareAmount = if (random.nextBoolean()) Some(5.0 + random.nextDouble() * 50.0) else None,
452      paymentType = if (random.nextBoolean()) Some(List("card", "cash", "mobile").apply(random.nextInt(3))) else None
453    )
454  }
455}

  1import os
  2import boto3
  3import pymysql
  4import redis
  5from elasticsearch import Elasticsearch
  6from datetime import datetime, timedelta
  7import json
  8import logging
  9
 10class TaxiDataStorage:
 11    """
 12    Unified interface for multiple storage systems in the taxi pipeline
 13    """
 14
 15    def __init__(self):
 16        self.mysql_client = self._create_mysql_client()
 17        self.dynamodb_client = self._create_dynamodb_client()
 18        self.s3_client = self._create_s3_client()
 19        self.redis_client = self._create_redis_client()
 20        self.es_client = self._create_elasticsearch_client()
 21        self.logger = self._setup_logging()
 22
 23    def _create_mysql_client(self):
 24        """Create MySQL connection for transactional data"""
 25        return pymysql.connect(
 26            host=os.environ['MYSQL_HOST'],
 27            port=int(os.environ.get('MYSQL_PORT', 3306)),
 28            user=os.environ['MYSQL_USER'],
 29            password=os.environ['MYSQL_PASSWORD'],
 30            database=os.environ['MYSQL_DATABASE'],
 31            charset='utf8mb4',
 32            autocommit=True,
 33            connect_timeout=30,
 34            read_timeout=30,
 35            write_timeout=30
 36        )
 37
 38    def _create_dynamodb_client(self):
 39        """Create DynamoDB client for real-time data"""
 40        return boto3.client(
 41            'dynamodb',
 42            region_name=os.environ.get('AWS_REGION', 'us-east-1'),
 43            aws_access_key_id=os.environ['AWS_ACCESS_KEY_ID'],
 44            aws_secret_access_key=os.environ['AWS_SECRET_ACCESS_KEY']
 45        )
 46
 47    def _create_s3_client(self):
 48        """Create S3 client for data lake storage"""
 49        return boto3.client(
 50            's3',
 51            region_name=os.environ.get('AWS_REGION', 'us-east-1'),
 52            aws_access_key_id=os.environ['AWS_ACCESS_KEY_ID'],
 53            aws_secret_access_key=os.environ['AWS_SECRET_ACCESS_KEY']
 54        )
 55
 56    def _create_redis_client(self):
 57        """Create Redis client for caching and real-time metrics"""
 58        return redis.Redis(
 59            host=os.environ['REDIS_HOST'],
 60            port=int(os.environ.get('REDIS_PORT', 6379)),
 61            password=os.environ.get('REDIS_PASSWORD'),
 62            decode_responses=True,
 63            socket_connect_timeout=10,
 64            socket_timeout=10,
 65            retry_on_timeout=True,
 66            health_check_interval=30
 67        )
 68
 69    def _create_elasticsearch_client(self):
 70        """Create Elasticsearch client for search and analytics"""
 71        return Elasticsearch(
 72            hosts=[{'host': os.environ['ES_HOST'], 'port': int(os.environ.get('ES_PORT', 9200))}],
 73            http_auth=(os.environ.get('ES_USERNAME'), os.environ.get('ES_PASSWORD')),
 74            use_ssl=os.environ.get('ES_USE_SSL', 'false').lower() == 'true',
 75            verify_certs=False,
 76            timeout=30,
 77            max_retries=3,
 78            retry_on_timeout=True
 79        )
 80
 81    def save_batch_analytics(self, analytics_data):
 82        """Save batch processing results to MySQL and Hive"""
 83        try:
 84            with self.mysql_client.cursor() as cursor:
 85                # Insert daily trip summary
 86                for record in analytics_data:
 87                    sql = """
 88                    INSERT INTO daily_trip_summary
 89                    (date, pickup_zone, total_trips, total_revenue, avg_fare,
 90                     avg_distance, avg_duration, total_tips, active_vendors)
 91                    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
 92                    ON DUPLICATE KEY UPDATE
 93                    total_trips = VALUES(total_trips),
 94                    total_revenue = VALUES(total_revenue),
 95                    avg_fare = VALUES(avg_fare),
 96                    avg_distance = VALUES(avg_distance),
 97                    avg_duration = VALUES(avg_duration),
 98                    total_tips = VALUES(total_tips),
 99                    active_vendors = VALUES(active_vendors),
100                    updated_at = CURRENT_TIMESTAMP
101                    """
102
103                    cursor.execute(sql, (
104                        record['date'],
105                        record['pickup_zone'],
106                        record['total_trips'],
107                        record['total_revenue'],
108                        record['avg_fare'],
109                        record['avg_distance'],
110                        record['avg_duration'],
111                        record['total_tips'],
112                        record['active_vendors']
113                    ))
114
115            self.logger.info(f"Saved {len(analytics_data)} batch analytics records to MySQL")
116
117        except Exception as e:
118            self.logger.error(f"Failed to save batch analytics: {str(e)}")
119            raise
120
121    def save_realtime_event(self, event_data):
122        """Save real-time taxi events to DynamoDB"""
123        try:
124            item = {
125                'trip_id': {'S': event_data['trip_id']},
126                'timestamp': {'N': str(event_data['timestamp'])},
127                'event_type': {'S': event_data['event_type']},
128                'vendor_id': {'N': str(event_data['vendor_id'])},
129                'latitude': {'N': str(event_data['latitude'])},
130                'longitude': {'N': str(event_data['longitude'])},
131                'zone': {'S': event_data.get('zone', 'unknown')},
132                'passenger_count': {'N': str(event_data['passenger_count'])},
133                'processing_time': {'S': datetime.utcnow().isoformat()}
134            }
135
136            if event_data.get('fare_amount'):
137                item['fare_amount'] = {'N': str(event_data['fare_amount'])}
138
139            if event_data.get('payment_type'):
140                item['payment_type'] = {'S': event_data['payment_type']}
141
142            response = self.dynamodb_client.put_item(
143                TableName='taxi_events_realtime',
144                Item=item
145            )
146
147            self.logger.debug(f"Saved event {event_data['trip_id']} to DynamoDB")
148            return response
149
150        except Exception as e:
151            self.logger.error(f"Failed to save real-time event: {str(e)}")
152            raise
153
154    def cache_realtime_metrics(self, metrics_data):
155        """Cache real-time metrics in Redis for dashboard consumption"""
156        try:
157            pipeline = self.redis_client.pipeline()
158            current_time = int(datetime.utcnow().timestamp())
159
160            for zone, metrics in metrics_data.items():
161                key_prefix = f"realtime:{zone}"
162
163                # Cache trip counts (expire in 5 minutes)
164                pipeline.setex(f"{key_prefix}:trip_count", 300, metrics['trip_count'])
165                pipeline.setex(f"{key_prefix}:avg_fare", 300, metrics['avg_fare'])
166                pipeline.setex(f"{key_prefix}:active_drivers", 300, metrics['active_drivers'])
167
168                # Add to time series for trending (keep 24 hours)
169                pipeline.zadd(f"{key_prefix}:trip_series", {current_time: metrics['trip_count']})
170                pipeline.zremrangebyscore(f"{key_prefix}:trip_series", 0, current_time - 86400)
171
172                # Set expiration on time series keys
173                pipeline.expire(f"{key_prefix}:trip_series", 86400)
174
175            pipeline.execute()
176            self.logger.info(f"Cached metrics for {len(metrics_data)} zones")
177
178        except Exception as e:
179            self.logger.error(f"Failed to cache realtime metrics: {str(e)}")
180            raise
181
182    def index_for_search(self, trip_data):
183        """Index trip data in Elasticsearch for search and analytics"""
184        try:
185            actions = []
186            for trip in trip_data:
187                doc = {
188                    '_index': f"taxi-trips-{datetime.now().strftime('%Y-%m')}",
189                    '_type': '_doc',
190                    '_id': trip['trip_id'],
191                    '_source': {
192                        'trip_id': trip['trip_id'],
193                        'vendor_id': trip['vendor_id'],
194                        'pickup_datetime': trip['pickup_datetime'],
195                        'dropoff_datetime': trip.get('dropoff_datetime'),
196                        'pickup_location': {
197                            'lat': trip['pickup_latitude'],
198                            'lon': trip['pickup_longitude']
199                        },
200                        'dropoff_location': {
201                            'lat': trip.get('dropoff_latitude'),
202                            'lon': trip.get('dropoff_longitude')
203                        } if trip.get('dropoff_latitude') else None,
204                        'pickup_zone': trip['pickup_zone'],
205                        'dropoff_zone': trip.get('dropoff_zone'),
206                        'passenger_count': trip['passenger_count'],
207                        'trip_distance': trip.get('trip_distance'),
208                        'fare_amount': trip.get('fare_amount'),
209                        'tip_amount': trip.get('tip_amount'),
210                        'total_amount': trip.get('total_amount'),
211                        'payment_type': trip.get('payment_type'),
212                        'indexed_at': datetime.utcnow().isoformat()
213                    }
214                }
215                actions.append(doc)
216
217            # Bulk index documents
218            from elasticsearch.helpers import bulk
219            bulk(self.es_client, actions)
220            self.logger.info(f"Indexed {len(actions)} documents in Elasticsearch")
221
222        except Exception as e:
223            self.logger.error(f"Failed to index documents: {str(e)}")
224            raise
225
226    def archive_to_s3(self, data, s3_bucket, s3_key):
227        """Archive processed data to S3 data lake"""
228        try:
229            # Convert data to JSON lines format
230            json_data = '\n'.join([json.dumps(record) for record in data])
231
232            # Upload to S3 with compression
233            self.s3_client.put_object(
234                Bucket=s3_bucket,
235                Key=s3_key,
236                Body=json_data.encode('utf-8'),
237                ContentEncoding='gzip',
238                StorageClass='STANDARD_IA',  # Infrequent access for archival
239                Metadata={
240                    'created_at': datetime.utcnow().isoformat(),
241                    'record_count': str(len(data)),
242                    'pipeline_version': '1.0'
243                }
244            )
245
246            self.logger.info(f"Archived {len(data)} records to s3://{s3_bucket}/{s3_key}")
247
248        except Exception as e:
249            self.logger.error(f"Failed to archive to S3: {str(e)}")
250            raise
251
252    def get_realtime_zone_metrics(self, zone):
253        """Retrieve real-time metrics for a specific zone"""
254        try:
255            key_prefix = f"realtime:{zone}"
256
257            metrics = {}
258            metrics['trip_count'] = self.redis_client.get(f"{key_prefix}:trip_count") or 0
259            metrics['avg_fare'] = float(self.redis_client.get(f"{key_prefix}:avg_fare") or 0)
260            metrics['active_drivers'] = self.redis_client.get(f"{key_prefix}:active_drivers") or 0
261
262            # Get time series data for trending
263            current_time = int(datetime.utcnow().timestamp())
264            hour_ago = current_time - 3600
265
266            trip_series = self.redis_client.zrangebyscore(
267                f"{key_prefix}:trip_series",
268                hour_ago,
269                current_time,
270                withscores=True
271            )
272
273            metrics['trip_trend'] = [(int(score), int(value)) for value, score in trip_series]
274
275            return metrics
276
277        except Exception as e:
278            self.logger.error(f"Failed to get realtime metrics for {zone}: {str(e)}")
279            return {}
280
281    def search_trips(self, query_params):
282        """Search trip data using Elasticsearch"""
283        try:
284            # Build Elasticsearch query
285            query = {
286                "query": {
287                    "bool": {
288                        "must": []
289                    }
290                },
291                "sort": [{"pickup_datetime": {"order": "desc"}}],
292                "size": query_params.get('limit', 100)
293            }
294
295            # Add filters based on query parameters
296            if query_params.get('zone'):
297                query["query"]["bool"]["must"].append({
298                    "term": {"pickup_zone": query_params['zone']}
299                })
300
301            if query_params.get('date_from'):
302                query["query"]["bool"]["must"].append({
303                    "range": {
304                        "pickup_datetime": {
305                            "gte": query_params['date_from']
306                        }
307                    }
308                })
309
310            if query_params.get('fare_min'):
311                query["query"]["bool"]["must"].append({
312                    "range": {
313                        "fare_amount": {
314                            "gte": query_params['fare_min']
315                        }
316                    }
317                })
318
319            # Geospatial search if coordinates provided
320            if query_params.get('lat') and query_params.get('lon'):
321                query["query"]["bool"]["must"].append({
322                    "geo_distance": {
323                        "distance": query_params.get('radius', '1km'),
324                        "pickup_location": {
325                            "lat": query_params['lat'],
326                            "lon": query_params['lon']
327                        }
328                    }
329                })
330
331            # Execute search
332            response = self.es_client.search(
333                index="taxi-trips-*",
334                body=query
335            )
336
337            results = []
338            for hit in response['hits']['hits']:
339                results.append(hit['_source'])
340
341            return {
342                'total': response['hits']['total']['value'],
343                'results': results
344            }
345
346        except Exception as e:
347            self.logger.error(f"Failed to search trips: {str(e)}")
348            return {'total': 0, 'results': []}
349
350    def get_analytics_summary(self, date_from, date_to):
351        """Get analytics summary from MySQL"""
352        try:
353            with self.mysql_client.cursor(pymysql.cursors.DictCursor) as cursor:
354                sql = """
355                SELECT
356                    pickup_zone,
357                    SUM(total_trips) as total_trips,
358                    SUM(total_revenue) as total_revenue,
359                    AVG(avg_fare) as avg_fare,
360                    AVG(avg_distance) as avg_distance,
361                    MAX(total_trips) as peak_trips,
362                    COUNT(DISTINCT date) as active_days
363                FROM daily_trip_summary
364                WHERE date BETWEEN %s AND %s
365                GROUP BY pickup_zone
366                ORDER BY total_revenue DESC
367                LIMIT 20
368                """
369
370                cursor.execute(sql, (date_from, date_to))
371                results = cursor.fetchall()
372
373                return results
374
375        except Exception as e:
376            self.logger.error(f"Failed to get analytics summary: {str(e)}")
377            return []
378
379    def _setup_logging(self):
380        """Setup logging configuration"""
381        logging.basicConfig(
382            level=logging.INFO,
383            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
384        )
385        return logging.getLogger(__name__)
386
387    def health_check(self):
388        """Perform health check on all storage systems"""
389        health_status = {}
390
391        # MySQL health check
392        try:
393            with self.mysql_client.cursor() as cursor:
394                cursor.execute("SELECT 1")
395                health_status['mysql'] = 'healthy'
396        except Exception as e:
397            health_status['mysql'] = f'unhealthy: {str(e)}'
398
399        # Redis health check
400        try:
401            self.redis_client.ping()
402            health_status['redis'] = 'healthy'
403        except Exception as e:
404            health_status['redis'] = f'unhealthy: {str(e)}'
405
406        # Elasticsearch health check
407        try:
408            self.es_client.cluster.health()
409            health_status['elasticsearch'] = 'healthy'
410        except Exception as e:
411            health_status['elasticsearch'] = f'unhealthy: {str(e)}'
412
413        # DynamoDB health check
414        try:
415            self.dynamodb_client.describe_table(TableName='taxi_events_realtime')
416            health_status['dynamodb'] = 'healthy'
417        except Exception as e:
418            health_status['dynamodb'] = f'unhealthy: {str(e)}'
419
420        return health_status
421
422    def cleanup_old_data(self):
423        """Cleanup old data based on retention policies"""
424        try:
425            current_date = datetime.utcnow()
426
427            # Clean up old Redis time series (keep 7 days)
428            week_ago = int((current_date - timedelta(days=7)).timestamp())
429            keys = self.redis_client.keys("realtime:*:trip_series")
430
431            pipeline = self.redis_client.pipeline()
432            for key in keys:
433                pipeline.zremrangebyscore(key, 0, week_ago)
434            pipeline.execute()
435
436            self.logger.info(f"Cleaned up {len(keys)} Redis time series")
437
438            # Archive and clean up old Elasticsearch indices (keep 6 months)
439            six_months_ago = current_date - timedelta(days=180)
440            old_index = f"taxi-trips-{six_months_ago.strftime('%Y-%m')}"
441
442            if self.es_client.indices.exists(index=old_index):
443                # Archive to S3 before deletion
444                self._archive_elasticsearch_index(old_index)
445                self.es_client.indices.delete(index=old_index)
446                self.logger.info(f"Archived and deleted old index: {old_index}")
447
448        except Exception as e:
449            self.logger.error(f"Failed to cleanup old data: {str(e)}")
450
451    def _archive_elasticsearch_index(self, index_name):
452        """Archive Elasticsearch index to S3 before deletion"""
453        # Implementation would use Elasticsearch's scroll API to export data
454        # and save to S3 in compressed format
455        pass
456
457
458# Example usage and configuration
459if __name__ == "__main__":
460    storage = TaxiDataStorage()
461
462    # Example batch analytics save
463    analytics_data = [
464        {
465            'date': '2024-01-01',
466            'pickup_zone': 'Manhattan_Midtown',
467            'total_trips': 5000,
468            'total_revenue': 125000.0,
469            'avg_fare': 25.0,
470            'avg_distance': 2.5,
471            'avg_duration': 15.0,
472            'total_tips': 20000.0,
473            'active_vendors': 150
474        }
475    ]
476
477    storage.save_batch_analytics(analytics_data)
478
479    # Example real-time event
480    event = {
481        'trip_id': 'trip_12345',
482        'timestamp': int(datetime.utcnow().timestamp() * 1000),
483        'event_type': 'pickup',
484        'vendor_id': 1,
485        'latitude': 40.7589,
486        'longitude': -73.9851,
487        'zone': 'Manhattan_Midtown',
488        'passenger_count': 2
489    }
490
491    storage.save_realtime_event(event)
492
493    # Health check
494    health = storage.health_check()
495    print(f"System health: {health}")