start implement streams primitives

mgoblin · Jan 4, 2018 · 209bf3f · 209bf3f
1 parent fe2b9a6
commit 209bf3f
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 9 deletions.
diff --git a/src/main/scala/ru/mg/Main.scala b/src/main/scala/ru/mg/Main.scala
@@ -3,8 +3,8 @@ package ru.mg
 import com.typesafe.scalalogging.LazyLogging
 import org.apache.flink.streaming.api.TimeCharacteristic
 import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
-import ru.mg.csv.PaymentsStream
 import ru.mg.detectors.Detectors._
+import ru.mg.streams.CsvFilePaymentsStream
 
 object Main extends LazyLogging {
   def main(args: Array[String]): Unit = {
@@ -13,7 +13,7 @@ object Main extends LazyLogging {
     val env = StreamExecutionEnvironment.getExecutionEnvironment
     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
 
-    val input = PaymentsStream(env, "data/payments.csv")
+    val input = CsvFilePaymentsStream(env, "data/payments.csv")
 
     frequentOutgoings(input)
       .addSink(s => logger.info(s"$s"))

diff --git a/...main/scala/ru/mg/csv/PaymentsStream.scala → ...ru/mg/streams/CsvFilePaymentsStream.scala b/...main/scala/ru/mg/csv/PaymentsStream.scala → ...ru/mg/streams/CsvFilePaymentsStream.scala
@@ -1,12 +1,12 @@
-package ru.mg.csv
+package ru.mg.streams
 
 import org.apache.flink.api.scala.createTypeInformation
 import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
+import ru.mg.csv.CsvStreamBuilder
 import ru.mg.domain.payment.Payment
-
 import ru.mg.domain.payment.Payments._
 
-object PaymentsStream {
+object CsvFilePaymentsStream {
   def apply(env: StreamExecutionEnvironment, filePath: String): DataStream[Payment] ={
     val csvStreamBuilder = new CsvStreamBuilder(filePath)
 

diff --git a/src/test/scala/ru/mg/detectors/FrequentOutgoingSpec.scala b/src/test/scala/ru/mg/detectors/FrequentOutgoingSpec.scala
@@ -9,10 +9,10 @@ import org.apache.flink.streaming.api.windowing.time.Time
 import org.junit.runner.RunWith
 import org.scalatest.junit.JUnitRunner
 import org.scalatest.{FlatSpec, Matchers}
-import ru.mg.csv.PaymentsStream
 import ru.mg.detectors.Detectors._
 import ru.mg.domain.fraud.Fraud
 import ru.mg.domain.payment.{Payment, Person}
+import ru.mg.streams.CsvFilePaymentsStream
 import ru.mg.utils.{FileUtils, SinkCollector}
 
 @RunWith(classOf[JUnitRunner])
@@ -23,7 +23,7 @@ class FrequentOutgoingSpec extends FlatSpec with Serializable with Matchers {
     val env = StreamExecutionEnvironment.createLocalEnvironment(parallelism = 1)
     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
 
-    val input = PaymentsStream(env, tempFile.getAbsolutePath)
+    val input = CsvFilePaymentsStream(env, tempFile.getAbsolutePath)
 
     SinkCollector.clear()
     val collector = SinkCollector[Fraud]
@@ -55,7 +55,7 @@ class FrequentOutgoingSpec extends FlatSpec with Serializable with Matchers {
     val env = StreamExecutionEnvironment.createLocalEnvironment(parallelism = 1)
     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
 
-    val input = PaymentsStream(env, tempFile.getAbsolutePath)
+    val input = CsvFilePaymentsStream(env, tempFile.getAbsolutePath)
 
     SinkCollector.clear()
     val collector = SinkCollector[Fraud]
@@ -76,7 +76,7 @@ class FrequentOutgoingSpec extends FlatSpec with Serializable with Matchers {
     val env = StreamExecutionEnvironment.createLocalEnvironment(parallelism = 1)
     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
 
-    val input = PaymentsStream(env, tempFile.getAbsolutePath)
+    val input = CsvFilePaymentsStream(env, tempFile.getAbsolutePath)
 
     SinkCollector.clear()
     val collector = SinkCollector[Fraud]