A place to discuss and ask questions about using Scala for Spark programming.
The flow was Server log adapter -> kafka -> flume -> hdfs for batch and long term. Kafka -> Spark Streaming -> Elasticsearch -> Grafana (for a nice dashboard)
Do u have a tutorial for building such pipeline please ? I wanna make the same workflow for my company
Hello! How to define a specific spark node to a partition key?
I need to run a similarity join algorithm of datasets in a heterogeneous cluster.
In The first step of the program, the records of the dataset are assigned signatures, and records with the same signature are sent to the same processing node.
Are set pairs sharing no signature cannot be similar. Thus, an intuitive approach is to use signatures as partition keys determining the workers to which a record is sent. Thus, only pairs with a signature in common are sent to the same worker avoiding unnecessary data transmission and similarity evaluations.
But, I need to ensure that signatures that have many records are sent to a worker with better processing power, for example, a GPU. So, how to define a specific spark node to a partition key?
According to this link delta-io/delta#526,
reservoidId is what I should look at. But I'm not able to figure out how to get the latest reservoirId from delta tables source location.
Upgrading our code from Spark 2.4.7 to 3.0.1, and getting this error: object typed in package scalalang is deprecated (since 3.0.0): please use untyped builtin aggregate functions in this code:
val meanSores: Dataset[(Int, Double)] = classified
.filter(e => e._2 >= 0.0 && e._2 <= 1.0)
.groupByKey { case (typeId, _) => typeId }
.agg(typed.avg(_._2))But I can’t figure out what it’s supposed to look like. All the examples I could find online use that typed.someAggregateFunction pattern...
The type of classified in this example is Dataset[(Int, Double)]
i create a smaller cluster with docker with one node master e 2 worker wint installed yarn hadoop 3.2 and spark 3.0. inside the docker all work fine. From spark shell i can work and make query to HIVE.
i try to run my application from intellij using the property -spark.master="docker:7070" the application run but no found properly HDFS and HIVE
how can configure correctly the run application ?
i don't understand why if i run from intellij spark application no read the server configuration
Is there a way to query data while the computations are being made in the workers, or really I have to load all the data I need in memory and do joins with it? Its a lot of big tables that need to be looked up to query data thats used inside the udf's, and to me it seems kinda of a waste to load it all in memory.
Need clarification on role of Catalyst optimizer and Project Tungsten
Catalyst optimizer will produce optimized Physical plan from logical plan.
The optimized physical plan will then taken by Code generator to emit Rdd's.
Here the Code generator is part of Project Tungsten or Catalyst Optimizer?
Here the Code generator is called as Whole Stage Code generator?
When I look at lot of article Code generator will emit Rdd.
Who will then generate bytecode from the RDD.
I am totally confused with Code generator and Project Tungsten. Please help me
A B expected
1 1 0
1 2 0
2 2 1
2 3 1
3 3 3
4 4 5
TL;DR: How to write ds.withWatermark("timestamp", "10 minutes").groupBy(window($"timestamp", "10 minutes", "5 minutes"), $"word").count() in a type-safe way ?
Hi all. I'm looking for Spark Structured Streaming, and I wonder if there is a way to use the window operations while sticking to the type-safe API.
Hi, since I did not yet get a reply anywhere else, please allow me the following question in this room.
Is it possible to run a Spark program (with Spark Context and Spark Session) from within an Akka Streams recipe for each element that passes a certain stage, let's say a mapAsync stage with parallelism set to 1?
mapAsync stage start the Spark Program, process its outcome wrapped by a Future when it terminates (successfully or failed), and then start it for the next Workspace ID that comes flowing through the stage.
if anyone can try the following in a Scala worksheet and tell me what they get, that would be incredibly helpful --
import org.apache.spark.sql.catalyst.parser._
import org.apache.spark.sql.internal.SQLConf
val sqlParser = new CatalystSqlParser(SQLConf.get)
val query = "select col1 from table1;"
//import sqlParser.astBuilder
val parsed = sqlParser.parseExpression(query)
//println(astBuilder.toString)
println(s"parsed: ${parsed.prettyJson}")the error I get looks absurd --
org.apache.spark.sql.catalyst.parser.ParseException:
mismatched input 'from' expecting {<EOF>, '-'}(line 1, pos 12)
== SQL ==
select col1 from table1;
------------^^^
at org.apache.spark.sql.catalyst.parser.ParseException.withCommand(ParseDriver.scala:266)
at org.apache.spark.sql.catalyst.parser.AbstractSqlParser.parse(ParseDriver.scala:133)
at org.apache.spark.sql.catalyst.parser.AbstractSqlParser.parseExpression(ParseDriver.scala:49)
... 36 elided
Collect in spark is taking huge time. I want to get list of values of one column to Scala collection. How can I do this? val newDynamicFieldTablesDF = cachedPhoenixAppMetaDataForCreateTableDF
.select(col("reporting_table")).except(clientSchemaDF)
logger.info(s"####### except with client-schema done " + LocalDateTime.now())
// newDynamicFieldTablesDF.cache()
val detailsForCreateTableDF = cachedPhoenixAppMetaDataForCreateTableDF
.join(broadcast(newDynamicFieldTablesDF), Seq("reporting_table"), "inner")
logger.info(s"####### join with newDF done " + LocalDateTime.now())
// detailsForCreateTableDF.cache()
val newDynamicFieldTablesList = newDynamicFieldTablesDF.map(r => r.getString(0)).collect().toSetnewDynamicFieldTablesList.foreach(table => {
// create string interpolation for create table goes here
val tableInformationDF = detailsForCreateTableDF.filter(col("reporting_table") === table)
tableInformationDF.rdd.collect().foreach(row => {
val reportingTable = row.getAs("reporting_table").asInstanceOf[String]
val dynamicFieldEntityId = row.getAs("dynamic_field_entity_id").asInstanceOf[String]
val isMultiSet = row.getAs("is_multi_set").asInstanceOf[Boolean]
val individualIdForMultiSet = row.getAs("individual_id_for_multiset").asInstanceOf[String]
val dynamicFieldColumns = row.getAs("dynamic_field_columns").asInstanceOf[String]
val textImageOn = row.getAs("text_image_on").asInstanceOf[String]
val dynamicFieldColumnsForCreate = if (isMultiSet) s"$individualIdForMultiSet $multiSetColumns" else ""
val entity = row.getAs("dynamic_field_entity").asInstanceOf[String]
val entityTableName = if (entity.equalsIgnoreCase("financial_allocation"))
"ENTITY_COST_CENTER_ALLOCATION" else entity
val entityId = row.getAs("dynamic_field_entity_id").asInstanceOf[String]
if (!table.toLowerCase.contains("custodian")) {
val createTableQuery =
s"""
|CREATE TABLE [dbo].[$reportingTable](
| [${reportingTable}_id] [int] NOT NULL,
| [${dynamicFieldEntityId.toLowerCase}] [int] NOT NULL,
| $dynamicFieldColumnsForCreate
| $dynamicFieldColumns
| CONSTRAINT [${reportingTable}_1] PRIMARY KEY CLUSTERED
| ([${reportingTable}_id] ASC) WITH (PAD_INDEX = OFF, IGNORE_DUP_KEY = OFF) ON [PRIMARY],
| CONSTRAINT [FK_$reportingTable] FOREIGN KEY ([$entityId])
| REFERENCES $entityTableName ($entityId)
| ON DELETE CASCADE
| ) ON
| [PRIMARY]
| $textImageOn
""".stripMargin
val dynamicFieldColForIndex = if (isMultiSet) " , [dynamic_field_data_set_id] ASC " else ""
val createIndexQuery =
s"""
|CREATE NONCLUSTERED INDEX [${reportingTable}_2] ON [dbo].[$reportingTable]
| ( [$dynamicFieldEntityId] ASC $dynamicFieldColForIndex
| )WITH (PAD_INDEX = OFF, SORT_IN_TEMPDB = OFF, DROP_EXISTING = OFF, IGNORE_DUP_KEY = OFF,
| ONLINE = OFF) ON [PRIMARY]
""".stripMargin