|
@@ -1,120 +0,0 @@
|
1
|
|
-package com.mediahub
|
2
|
|
-
|
3
|
|
-import akka.NotUsed
|
4
|
|
-import akka.actor.ActorSystem
|
5
|
|
-import akka.stream.ActorAttributes.supervisionStrategy
|
6
|
|
-import akka.stream.Supervision.resumingDecider
|
7
|
|
-import akka.stream.alpakka.csv.scaladsl.{CsvParsing, CsvToMap}
|
8
|
|
-import akka.stream.scaladsl.{Compression, FileIO, Flow, Sink, Source}
|
9
|
|
-import akka.util.ByteString
|
10
|
|
-import org.springframework.stereotype.Component
|
11
|
|
-
|
12
|
|
-import java.io.File
|
13
|
|
-import java.nio.file.Paths
|
14
|
|
-import scala.collection.immutable
|
15
|
|
-import scala.concurrent.duration.DurationInt
|
16
|
|
-import scala.concurrent.{Await, Future}
|
17
|
|
-import scala.language.postfixOps
|
18
|
|
-
|
19
|
|
-
|
20
|
|
-@Component
|
21
|
|
-class FileReader {
|
22
|
|
-
|
23
|
|
- implicit val system: ActorSystem = ActorSystem()
|
24
|
|
-
|
25
|
|
- val titleBasicsResource: File = new File("src/main/resources/title.basics.tsv.gz")
|
26
|
|
- val titlePrincipalsResource: File = new File("src/main/resources/title.principals.tsv.gz")
|
27
|
|
- val nameBasicsResource: File = new File("src/main/resources/name.basics.tsv.gz")
|
28
|
|
-
|
29
|
|
- def fileSource(file: File): Source[ByteString, NotUsed] = {
|
30
|
|
- Source.single(file)
|
31
|
|
- .flatMapConcat(f => FileIO.fromPath(Paths.get(f.getPath), 1 * 1024 * 1024 * 2))
|
32
|
|
- .via(Compression.gunzip())
|
33
|
|
- }
|
34
|
|
-
|
35
|
|
- val lineParser: Flow[ByteString, Map[String, String], NotUsed] = {
|
36
|
|
- CsvParsing
|
37
|
|
- .lineScanner(CsvParsing.Tab, CsvParsing.DoubleQuote, CsvParsing.DoubleQuote)
|
38
|
|
- .via(CsvToMap.toMapAsStringsCombineAll(headerPlaceholder = Option.empty))
|
39
|
|
- }
|
40
|
|
-
|
41
|
|
- //working search not optimized
|
42
|
|
- val titleId: String = getIdOfTitleAsync(titleBasicsResource, "Carmencita")
|
43
|
|
- println("value ", titleId)
|
44
|
|
-
|
45
|
|
- val personsIdList: List[String] = getIdOfPersonsAsync(titlePrincipalsResource, titleId).toList.flatten
|
46
|
|
- println("value ", personsIdList)
|
47
|
|
-
|
48
|
|
- val personsList = getPersons(nameBasicsResource, personsIdList)
|
49
|
|
- println("value ", personsList)
|
50
|
|
-
|
51
|
|
-
|
52
|
|
- def getIdOfTitleAsync(file: File, titleName: String): String = {
|
53
|
|
- val result = fileSource(file)
|
54
|
|
- .mapAsync(50) {
|
55
|
|
- res =>
|
56
|
|
- Source.single(res)
|
57
|
|
- .via(lineParser)
|
58
|
|
- .filter(row => row.getOrElse("primaryTitle", "") == titleName)
|
59
|
|
- .map(a => a.get("tconst"))
|
60
|
|
- .withAttributes(supervisionStrategy(resumingDecider))
|
61
|
|
- .runWith(Sink.head)
|
62
|
|
- }
|
63
|
|
- .withAttributes(supervisionStrategy(resumingDecider))
|
64
|
|
- .runWith(Sink.head)
|
65
|
|
-
|
66
|
|
- Await.result(result, 5 minutes)
|
67
|
|
- result.value.get.get.get
|
68
|
|
- }
|
69
|
|
-
|
70
|
|
- def getIdOfPersonsAsync(file: File, titleId: String): immutable.Iterable[Option[String]] = {
|
71
|
|
- val te = fileSource(file)
|
72
|
|
- .mapAsync(50) {
|
73
|
|
- res =>
|
74
|
|
- Source.single(res)
|
75
|
|
- .via(lineParser)
|
76
|
|
- .filter(row => row.getOrElse("tconst", "") == titleId)
|
77
|
|
- .map(a => a.get("nconst"))
|
78
|
|
- .withAttributes(supervisionStrategy(resumingDecider))
|
79
|
|
- .runWith(Sink.collection)
|
80
|
|
- }
|
81
|
|
- .withAttributes(supervisionStrategy(resumingDecider))
|
82
|
|
- .runWith(Sink.head)
|
83
|
|
-
|
84
|
|
- Await.result(te, 5 minutes)
|
85
|
|
- }
|
86
|
|
-
|
87
|
|
- def getPersons(file: File, personsIdList: List[String]): List[String] = {
|
88
|
|
- val result: Future[immutable.Iterable[Option[String]]] = Source.single(file)
|
89
|
|
- .flatMapConcat(f => FileIO.fromPath(Paths.get(f.getPath), 1 * 1024 * 1024))
|
90
|
|
- .via(Compression.gunzip())
|
91
|
|
- .via(CsvParsing.lineScanner(CsvParsing.Tab, CsvParsing.DoubleQuote, CsvParsing.DoubleQuote))
|
92
|
|
- .via(CsvToMap.toMapAsStringsCombineAll(headerPlaceholder = Option.empty))
|
93
|
|
- .filter(row => personsIdList.contains(row.getOrElse("nconst", "")))
|
94
|
|
- .map(a => a.get("primaryName"))
|
95
|
|
- .runWith(Sink.collection)
|
96
|
|
-
|
97
|
|
- Await.result(result, 5 minutes)
|
98
|
|
- result.value.get.get.toList.flatten
|
99
|
|
- }
|
100
|
|
-
|
101
|
|
- /* def getPersonsAsync(file: File, personsIdList: List[String]) = {
|
102
|
|
- val test = Source.single(file)
|
103
|
|
- .flatMapConcat(f => FileIO.fromPath(Paths.get(f.getPath), 1 * 1024 * 1024))
|
104
|
|
- .via(Compression.gunzip())
|
105
|
|
- .mapAsync(2) {
|
106
|
|
- res =>
|
107
|
|
- Source.single(res)
|
108
|
|
- .via(CsvParsing.lineScanner(CsvParsing.Tab, CsvParsing.DoubleQuote, CsvParsing.DoubleQuote))
|
109
|
|
- .via(CsvToMap.toMapAsStringsCombineAll(headerPlaceholder = Option.empty))
|
110
|
|
- .filter(row => row.getOrElse("nconst", "") == personsIdList.head)
|
111
|
|
- .map(a => a.get("primaryName"))
|
112
|
|
- .runWith(Sink.collection)
|
113
|
|
- }
|
114
|
|
- .runWith(Sink.collection)
|
115
|
|
-
|
116
|
|
- Await.result(test, 5 minutes)
|
117
|
|
- test.value.get.get.toList.flatten
|
118
|
|
- }
|
119
|
|
- */
|
120
|
|
-}
|