Mgiyuk

Is using function in transformation causing Not Serializable exceptions? I have a Breeze DenseMatrix , i find mean per row and mean of squares per row and put them in another DenseMatrix , one per column. But i get Task Not Serializable exception. I know that sc is not Serializable but i think that the exception is because i call functions in a transformation in Safe Zones. Breeze DenseMatrix mean mean DenseMatrix Task Not Serializable sc Serializable Am i right? And how could be a possible way to be done without any functions? Any help would be great! Code: object MotitorDetection { case class MonDetect() extends Serializable { var sc: SparkContext = _ var machines: Int=0 var counters: Int=0 var GlobalVec= BDM.zeros[Double](counters, 2) def findMean(a: BDM[Double]): BDV[Double] = { var c = mean(a(*, ::)) c} def toMatrix(x: BDV[Double], y: BDV[Double], C: Int): BDM[Double]={ val m = BDM.zeros[Double](C,2) m(::, 0) := x m(::, 1) := y m} def SafeZones(stream: DStream[(In...

Spark SQL nested query I have the following spark sql query SELECT count(*), channel FROM channelusage a WHERE a.starttime>= windowstarttime AND a.endtime <= windowendtime GROUP BY channel I have to generate these counts for 10 windows. Currently I use a while loop to generate the windowstarttime and windowendtime. What I want to do is - I want to generate the windows in the sql query itself something like a nested sql similar to this - SELECT count(*), channel FROM channelusage a WHERE (nested query logic) GROUP BY channel so that I get a output similar to this windowstarttime | windowentime | channel | count 11:00:01 | 11:00:10 | ABC |2 11:00:11 | 11:00:20 | ABC |4 11:00:21 | 11:00:30 | NBC |10 11:00:31 | 11:00:40 | CNN |5 What is your input data set? – Kannan Kandasamy Jun 29 at 20:24 ...

Best strategy for repartionBy with few big partitions I have to repartition geo data by quadkey. Primarily all the data is pretty balanced, but few partitions are 500x times bigger than others. So it causes very unbalanced partition stage, like 20-30 of 3500 tasks are 98 % slower than others. Is there are any good strategy in that case? I need to do next: stage.repartition(partitionColumns.map(new org.apache.spark.sql.Column(_)):_*) .write.partitionBy(partitionColumns:_*) .format("parquet") .option("compression", "gzip") .mode(SaveMode.Append) .save(destUrl) 1 Answer 1 The .repartition is unnecessary and is probably causing the issue. .repartition If you leave that out and just have the .write.partitionBy... , you will still get the same directory structure, you will just have multiple files within each directory. .write.partitionBy... ...

Spark cosmosdb connector and saveToCosmosDB return failed documents saveToCosmosDB api in spark to cosmos db connector does not have any return type. how do we get failed to ingest record in cosmos db? saveToCosmosDB api db return By clicking "Post Your Answer", you acknowledge that you have read our updated terms of service, privacy policy and cookie policy, and that your continued use of the website is subject to these policies.

Spark How to Specify Number of Resulting Files for DataFrame While/After Writing I saw several q/a's about writing single file into hdfs,it seems using coalesce(1) is sufficient. coalesce(1) E.g; df.coalesce(1).write.mode("overwrite").format(format).save(location) But how can I specify "exact" number of files that will written after save operation? So my question is; If I have dataframe which consist 100 partitions when I make write operation will it write 100 files? If I have dataframe which consist 100 partitions when I make write operation after calling repartition(50)/coalsesce(50) will it write 50 files? repartition(50)/coalsesce(50) Is there a way in spark which will allow to specify resulting number of files while writing dataframe into HDFS ? Thanks 1 Answer 1 Number of output files is in general equal to the number of writing tasks (partitions). Under normal condit...

Spark writing to Cassandra with varying TTL In Java Spark, I have a dataframe that has a 'bucket_timestamp' column, which represents the time of the bucket that the row belongs to. I want to write the dataframe to a Cassandra DB. The data must be written to the DB with TTL. The TTL should be depended on the bucket timestamp - where each row's TTL should be calculated as ROW_TTL = CONST_TTL - (CurrentTime - bucket_timestamp) , where CONST_TTL is a constant TTL that I configured. ROW_TTL = CONST_TTL - (CurrentTime - bucket_timestamp) CONST_TTL Currently I am writing to Cassandra with spark using a constant TTL, with the following code: df.write().format("org.apache.spark.sql.cassandra") .options(new HashMap<String, String>() { { put("keyspace", "key_space_name"); put("table, "table_name"); put("spark.cassandra.output.ttl, Long.toStri...

This summary is not available. Please click here to view the post.

How to exclude a jar from oozie user lib path I have two actions in an oozie job. One spark and one java action. <workflow-app name="batch-publisher" xmlns="uri:oozie:workflow:0.5"> <start to="spark-a5b5"/> <kill name="Kill"> <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message> </kill> <action name="spark-a5b5"> <spark xmlns="uri:oozie:spark-action:0.1"> <job-tracker>${jobTracker}</job-tracker> <name-node>${nameNode}</name-node> <configuration> <property> <name>oozie.libpath</name> <value>${nameNode}/user/hdfs/sps-batch-bi/Jars/</value> </property> <property> <name>oozie.use.system.libpath</na...