Bigdata ideas

When we use Futures in scala sometimes we need to combine sequentially and parallel execution of Futures. My research in this way is beginning from a task - generate a lot of Users (100K) and insert it all in different databases (Postgres and Cassandra) to make common score of write speed. case class User(id: Long, name: String, email: String, edomain: String) For Postgres, I should have used  Slick library and for Cassandra  Phantom  library (Schema safe, type-safe, reactive Scala driver for Cassandra/Datastax Enterprise). Steps ware easy, generate Seq of Users and in loop insert there into a database. With Postgres driver and with this count of items it was ok, every row was inserted without error. But Phantom is reactive and insert command return Future. (somewhere about) A snippet of code: abstract class CassUsers extends Table[CassUsers, User] with RootConnector { override def tableName: String = "us...

#1. Why method generates an error at compile time : Error here because the expression is expanding into SOMETHING match { case xs :Seq[Int] if xs.isEmpty => 0 case xs :Seq[Int] if xs.nonEmpty => xs.head + recursiveSum(xs.tail: _*) } and scala compiler doesn't know what is SOMETHING and his type. You think that it is xs : Int* parameter, but why. Functions definition can be like this def recursiveSum(a :Int,s :String, xs :Int*) :Int = { We can solve this by explicitly adding the parameter in the patter match, like this ( inside {} we use xs but it can be any name, it's not referred to internal parameter xs. ) def recursiveSum(xs :Int*) :Int = { xs match { case xs if xs.isEmpty => 0 case xs if xs.nonEmpty => xs.head + recursiveSum(xs.tail: _*) } } recursiveSum((1 to 5): _ *) result : 15 Also, we can do it without pattern matching, with IF val g: (Seq[Int]) => Int = si => { if (si.isEmpty) 0 else si.head + g(s...

In this topic, I will collect different questions and small tasks of Scala. Solutions or explanations will be on any page. All related by # - numbers. #1. Why method generates an error at compile time: error: missing parameter type for expanded function <console>:11: error: missing parameter type for expanded function The argument types of an anonymous function must be fully known. (SLS 8.5) Expected type was: Int        def recursiveSum(xs :Int*) :Int = {                                                               ^ For example, you write a method like this, to calculate sum recursively. def recursiveSum(xs :Int*) :Int = { case xs :Seq[Int] if xs.isEmpty => 0 case xs :Seq[Int] if xs.nonEmpty => xs.head + recursiveSum(xs.tail: _*) } #2. How to make a chain of PartialFunctions,...

In this short article, I will show a comparison of sequential and parallel executions with scala Futures. First of all, I have a sequence of tickers (Ticker) with a structure like this: case class Ticker(tickerId :Int, tickerCode :String, tickerFirst :String, tickerSeconds :String) I read it from Cassandra DB and it has size 28. val sess = Global.sessInstance val tickersFrx: Seq[Ticker] = sess.tickersDict println("tickersFrx.size = "+tickersFrx.size) Next, I need for each Ticker in this sequence execute 2 additional DB query. First to determine the maximum data date. And with knowledge of this date, I need to get maximum UNIX timestamp of last data tick. Producing Seq[TickerWithDdateTs] p.s. I use limit 1 because we have clustering keys ordered descending. select ddate from mts_src.ticks_count_days where ticker_id = :tickerId limit 1 select db_tsunx from mts_src.ticks where ticker_id = :tickerId and ddate= :pDdate limit 1 case class TickerWithDd...

In this short topic, I will show have run the calculation in parallel with the same code (method) but with individual sleep intervals. Basic - https://github.com/AlexGruPerm/barcl Formulation of the problem: 1) We have a flow of tick data from Forex, Stock market or etc (flow presented as a Column Family in Cassandra database). 2) And we have a couple of dictionaries:    dictionary of tickers (symbols) (ID, NAME), like 1-EURUSD, 2-ORCL.    dictionary of BWS (Bar width seconds) (TICKER_ID, BWS)- For example: 1 - 30 1 - 600 1 - 3600 It means that for ticker with ID=1 we need to calculate bars with widths 30 sec, 1 min, 1 hour( 60 mins). We need to calculate all bars online, with minimal lag between last tick and TS_END (last bar of ticker and ts_end as unixtimestamp ) of the last bar. Also, we need to eliminate unnecessary read of the database to minimize resource consumption. First, simple and unoptimized version was the next: in loop read all for...

In this topic, I will explain one example of integration akka-http (server-side) with Spark (spark session). Why it can be necessary sometimes? Imagine that: 1) You have source data (from Cassandra, Oracle, Parquet, etc) 2) And you have a machine learning model that uses this data to make classification, for example, Multilayer perceptron classifier (MLPC). MLPC is a classifier based on the feedforward artificial neural network. 3) You have written code on Scala for Spark that load source data, train MLPC model and can be used to predict output value (label) by input value(features). You want to run it all on Spark with standalone jar application and communicate with application from external, it can be RPC or any. I will use the easiest way - simple HTTP and HTML. 1) Source data description: I have Cassandra NoSQL database with previously calculated Column Family (like a table in RBDMS). Screenshot made from DataStax DevCenter 1.5.0 Here we have different Forex tickers inf...

I have one single instance Cassandra with source data, only 3 CF: There are 3 tables, mts_src.ticks - contains information about Forex tick data for 28 tickers, db_tsunx - Unixtimestamp. mts_src.ticks_count_days - Total ticks count by day and ticker_id. mts_src.ticks_count_total - Total ticks count by ticker_id. DROP TABLE mts_src.ticks; DROP TABLE mts_src.ticks_count_days; DROP TABLE mts_src.ticks_count_total; CREATE TABLE mts_src.ticks( ticker_id int , ddate date , ts bigint, db_tsunx bigint, ask double , bid double , PRIMARY KEY (( ticker_id, ddate ), ts, db_tsunx) ) WITH CLUSTERING ORDER BY ( ts DESC , db_tsunx DESC ); CREATE TABLE mts_src.ticks_count_days( ticker_id int , ddate date , ticks_count counter, PRIMARY KEY (ticker_id, ddate) ) WITH CLUSTERING ORDER BY ( ddate DESC ); CREATE TABLE mts_src.ticks_count_total( ticker_id int , ticks_count counter, PRIMARY KEY (ticker_id) ); Data from mts_src.ticks is using for bar...