Eduardo Rost eduardorost
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import org.apache.spark.SparkConf | |
| import org.apache.spark.sql.SparkSession | |
| import org.apache.spark.sql.types._ | |
| import org.slf4j.{Logger, LoggerFactory} | |
| object Main { | |
| val logger: Logger = LoggerFactory.getLogger(this.getClass) | |
| private lazy val sparkConf: SparkConf = new SparkConf() | |
| .setMaster("local[*]") |
eduardorost
/ streaming.mongodb.build.sbt
Last active
September 1, 2019 02:36
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| name := "streaming_mongodb" | |
| version := "0.1" | |
| scalaVersion := "2.12.9" | |
| libraryDependencies ++= Seq( | |
| "org.mongodb.scala" %% "mongo-scala-driver" % "2.7.0", | |
| "org.apache.spark" %% "spark-streaming" % "2.4.3", | |
| "org.apache.spark" %% "spark-sql" % "2.4.3", | |
| "org.apache.hadoop" % "hadoop-aws" % "3.2.0", | |
| "org.apache.hadoop" % "hadoop-common" % "3.2.0", |
eduardorost
/ spark.scala
Created
September 1, 2019 02:14
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| val sparkConf: SparkConf = new SparkConf() | |
| .setMaster("local[*]") | |
| .set("spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version", "2") | |
| .set("spark.streaming.driver.writeAheadLog.closeFileAfterWrite", "true") | |
| .set("spark.streaming.receiver.writeAheadLog.closeFileAfterWrite", "true") | |
| val spark: SparkSession = SparkSession | |
| .builder() | |
| .config(sparkConf) | |
| .getOrCreate() |
eduardorost
/ change.mongodb.scala
Last active
September 1, 2019 01:38
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| def process(changeDocument: ChangeStreamDocument[Document]) = { | |
| implicit val formats: DefaultFormats.type = DefaultFormats | |
| val rdd = spark.sparkContext.makeRDD(List(write(mapRecord(changeDocument)))) | |
| if (rddQueue.isEmpty) | |
| rddQueue.enqueue(rdd) | |
| else | |
| rddQueue.enqueue(rddQueue.dequeue().union(rdd)) | |
| } |
eduardorost
/ spark.send.s3.scala
Last active
September 1, 2019 01:16
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| def sendS3(rdd: RDD[String]) = { | |
| spark.read.json(rdd) | |
| .coalesce(1) | |
| .write | |
| .mode("append") | |
| .option("compression", "snappy") | |
| .parquet("s3://bucket/folder") | |
| } |
eduardorost
/ spark.streaming.queue.scala
Last active
September 1, 2019 01:16
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| val ssc = new StreamingContext(sparkConf, Seconds(60)) | |
| val rddQueue: collection.mutable.Queue[RDD[String]] = collection.mutable.Queue() | |
| ssc.queueStream(rddQueue) | |
| .foreachRDD(rdd => { | |
| if (!rdd.isEmpty()) sendS3(rdd) | |
| }) | |
| ssc.start() |
eduardorost
/ mongodb.watch.scala
Last active
September 1, 2019 02:18
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| val client = MongoClient.apply("mongodb.uri") | |
| val database = client.getDatabase("mongodb.database") | |
| val collection = database.getCollection("mongodb.collection") | |
| collection.watch() | |
| .fullDocument(FullDocument.UPDATE_LOOKUP) | |
| .subscribe(new Observer[ChangeStreamDocument[Document]] { | |
| override def onNext(changeDocument: ChangeStreamDocument[Document]): Unit = process(changeDocument) | |
| override def onError(e: Throwable): Unit = println(s"Error: $e") | |
| override def onComplete(): Unit = println("Completed") |
eduardorost
/ apache.conf
Last active
November 26, 2018 22:00
docker-compose and configuration for elastic stack
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| input | |
| { | |
| beats { | |
| port => 5044 | |
| } | |
| } | |
| filter | |
| { |