| def main(args: Array[String]): Unit = { | |
| val sc = SparkSession | |
| .builder() | |
| .appName(JOB_NAME) | |
| .getOrCreate() | |
| .sparkContext | |
| val ssc = new StreamingContext(sc, Seconds(BATCH_DURATION_SECONDS)) |
| package com.outbrain.datainfra.jester; | |
| import org.apache.commons.logging.Log; | |
| import org.apache.commons.logging.LogFactory; | |
| import org.apache.hadoop.classification.InterfaceAudience; | |
| import org.apache.hadoop.classification.InterfaceAudience.Private; | |
| import org.apache.hadoop.classification.InterfaceStability; | |
| import org.apache.hadoop.fs.FileStatus; | |
| import org.apache.hadoop.fs.FileSystem; | |
| import org.apache.hadoop.fs.Path; |
| val scalacOptions = Seq( | |
| "-encoding", | |
| "utf-8", // Specify character encoding used by source files. | |
| "-Ybackend-parallelism", // | |
| "8", | |
| "-explaintypes", // Explain type errors in more detail. | |
| "-feature", // Emit warning and location for usages of features that should be imported explicitly. | |
| "-unchecked", // Enable additional warnings where generated code depends on assumptions. | |
| "-Xcheckinit", // Wrap field accessors to throw an exception on uninitialized access. | |
| // "-Ymacro-annotations", // Enable support for macro annotations, formerly in macro paradise. |
| val groupId = "demo-consumer" | |
| val bootstrapServers = "localhost:9092" | |
| val spark = SparkSession | |
| .builder() | |
| .master("local[*]") | |
| .appName("demo") | |
| //.config("spark.sql.streaming.stateStore.maintenanceInterval", "10s") | |
| //.config("spark.sql.streaming.metricsEnabled", true) | |
| .config("spark.sql.streaming.minBatchesToRetain", 2) |
Once in a while, you may need to cleanup resources (containers, volumes, images, networks) ...
// see: https://github.com/chadoe/docker-cleanup-volumes
$ docker volume rm $(docker volume ls -qf dangling=true)
$ docker volume ls -qf dangling=true | xargs -r docker volume rm