import scala.collection.Map
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{Row, SQLContext, SaveMode}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.hive.HiveContext
object dataingest{
def main(args: Array[String]) {
def procIngestBAMData(url:String, username:String, passwd:String, qrySql:String ,hivetbl:String ,sqlContext: SQLContext) : Unit = {
println("URL : " + url)
println("username : " + username)
println("pwd : " + passwd)
println("sql : " + qrySql)
println("hivetable : " + hivetbl)
val driver = "com.microsoft.sqlserver.jdbc.SQLServerDriver"
val dfDynSrc = sqlContext.read.format("jdbc").option("url", url.trim).option("user", username.trim).option("password", passwd.trim).option("dbtable", qrySql).option("driver", driver).load() }
val driver = "com.microsoft.sqlserver.jdbc.SQLServerDriver"
var getDriverDataDF = spark.read.
format("jdbc").
option("url", s_url.trim).
option("user", s_username.trim).
option("password", s_passwd.trim).
option("dbtable", getdriverfileSQL).
option("driver", driver).
load()
val driverDatatoRDD = getDriverDataDF.rdd
if (driverDatatoRDD.count > 1 ) {
for (e <- driverDatatoRDD) {
val a = e(0)
val b = e(1)
val p_ActivityName = a.asInstanceOf[String]
val p_TableName = b.asInstanceOf[String]
log_message("procData processing names : " + p_ActivityName + " .../" + p_TableName )
val strtblName = p_TableName.toLowerCase
val strActivityName = p_ActivityName.toLowerCase
println("Activity name to process000 ====> " + strtblName )
println("Table Name00000 "+ strActivityName )
val sc2 = spark.sqlContext
procIngestBAMData(p_url,p_username , p_passwd, p_qry2, p_hivetbl, sc2) } else {
log_message("No records found for the given date range " + process_date + "." )
}
}
}
scala> val i: Integer = 33
i: Integer = 33
scala> i.asInstanceOf[Long]
java.lang.ClassCastException: java.lang.Integer cannot be cast to java.lang.Long
at scala.runtime.BoxesRunTime.unboxToLong(BoxesRunTime.java:110)
at $iwCiwCiwCiwC
KaTeX parse error: Unexpected character: '$' at position 36: 29)
at ̲$iwC: iwC.<init>(<console>:29)
at $iwCKaTeX parse error: Unexpected character: '$' at position 36: 34)
at ̲$iwC: iwC.<init>(<console>:34)
at $iwCat $iwCiwCiwC.<init>(<console>:38)
at $iwCiwC
KaTeX parse error: Unexpected character: '$' at position 36: 40)
at ̲$iwC: iwC.<init>(<console>:40)
at $iwCKaTeX parse error: Unexpected character: '$' at position 36: 42)
at ̲$iwC: iwC.<init>(<console>:42)
at $iwCat $iwC.<init>(<console>:46)
at <init>(<console>:48)
at .<init>(<console>:52)
at .<clinit>(<console>)
at .<init>(<console>:7)
at .<clinit>(<console>)
at $print(<console>)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.spark.repl.SparkIMain$ReadEvalPrint.call(SparkIMain.scala:1065)
at org.apache.spark.repl.SparkIMain$Request.loadAndRun(SparkIMain.scala:1346)
at org.apache.spark.repl.SparkIMain.loadAndRunReq$1(SparkIMain.scala:840)
at org.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:871)
at org.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:819)
at org.apache.spark.repl.SparkILoop.reallyInterpret$1(SparkILoop.scala:857)
at org.apache.spark.repl.SparkILoop.interpretStartingWith(SparkILoop.scala:902)
at org.apache.spark.repl.SparkILoop.command(SparkILoop.scala:814)
at org.apache.spark.repl.SparkILoop.processLine$1(SparkILoop.scala:657)
at org.apache.spark.repl.SparkILoop.innerLoop$1(SparkILoop.scala:665)
at org.apache.spark.repl.SparkILoop.org$apache$spark$repl$SparkILoopanonfun$org$apache$spark$repl$SparkILoop
KaTeX parse error: Unexpected character: '$' at position 7: process̲$1.apply$mcZ$s: process$1.apply$mcZ$sp(SparkILoop.scala:997)
at org.apache.spark.repl.SparkILoopKaTeX parse error: Unexpected character: '$' at position 7: process̲$1.apply(Spark: process$1.apply(SparkILoop.scala:945)
at org.apache.spark.repl.SparkILoopKaTeX parse error: Unexpected character: '$' at position 7: process̲$1.apply(Spark: process$1.apply(SparkILoop.scala:945)
at scala.tools.nsc.util.ScalaClassLoader$.savingContextLoader(ScalaClassLoader.scala:135)
at org.apache.spark.repl.SparkILoop.org$apache$spark$repl$SparkILoopat org.apache.spark.repl.SparkILoop.process(SparkILoop.scala:1059)
at org.apache.spark.repl.Main$.main(Main.scala:31)
at org.apache.spark.repl.Main.main(Main.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:731)
at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:181)
at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:206)
at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:121)
at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
<Income s"${income.`type`}Amount=${income.amount} ${income.`type`}Frequency=${income.frequency}"}"/>
Eg: if I pass int and string to the function as an input parameters, it should return true.
I have sort function defined as below:
case class Country(name: String, id: Int)
def sortT(compare: (T, T) => Boolean): List[T] = {
}
My question is, How do I sort the list of Country based on the name?
and How do I call this function having multiple parameter lists?
case class Country(name: String, id: Int)
def sort(list: List[T])(compare: (T, T) => Boolean): List[T] = {
}
In Spark I am not able to filter by existing column:Exception in thread "main" org.apache.spark.sql.AnalysisException: cannot resolve 'Inv. Pty' given input columns: [ο»ΏPstng Date, Name 1, Inv. Pty, Year]
Through Inv. Pty I am trying to filter. This column already in the dataframe input file you can see inside []
Here I have used dense_rank() as the analytical function, I want to pass this function as a argument incase if I want to pass any other function tomorrow,
val indexLatName = df.withColumn("LatestNameOrder", dense_rank()
.over(Window.partitionBy("`" + partBy + "`")
.orderBy(desc("`" + ordBy + "`"))))
val winFunc = "dense_rank"
Hiπ Everyone, I am new to Scala. I am struggling with a existing piece of scala code so need your help.
We (i.e. in my org) are using Gatling for perf testing. So we are using following piece of code:
ScenarioBuilder =
scenario(scn).during(config.duration) {
exec(session => {
val daysFromNow = nextInt(180) + 1 //avoid daysFromNow = 0 because current date is invalid for some locale half a world away
val siteAndLang = chooseFrom(siteLangs)
session.set("siteId", siteAndLang.substring(0, siteAndLang.indexOf("/")))
.set("locale", siteAndLang.substring(siteAndLang.indexOf("/")+1))
.set("hotelId", hotelIds)
.set("startDate", computeDateString(daysFromNow))
.set("endDate", computeDateString(daysFromNow + lengthOfStay(nextInt(lengthOfStay.length))))
.set("regionId", regionIds)
.set("adults", chooseFrom(adultCounts))
}).exec(addCookie(Cookie("DUAID", DUAID)))
.exec(http(scn)
.get(s"$endpoint/$path")
.header("Trace-ID", session => getTraceID())
.signatureCalculator(new CommonSignatureCalculator(DUAID))
.check(status is 200)
.check(validator)
)
}
ββββββββββββββββββ-ββββββββββββββββββ-ββββββββββββββββββ-
This is what we are passing as validator:
private def validator = (jsonPath("$.hotelId").exists)
So API which this script is hitting start sending compressed data and then it get following error:
19:44:50 14:14:50.084 [gatling-http-thread-1-3] ERROR io.gatling.http.response.StringResponseBody$ - Response body is not valid UTF-8 bytes
My question - how to unzip GZip response before pass to validator.
import java.time.LocalDate
import java.time.format.DateTimeFormatter._
import java.util.UUID
import config.RunConfiguration
import io.gatling.core.Predef._
import io.gatling.core.structure.ScenarioBuilder
import io.gatling.http.Predef._
import io.gatling.http.check.HttpCheck
import signature.CommonSignatureCalculator
import scala.io.Source
import scala.util.Random._
def buildScenario(
endpoint: String,
path: String,
scn: String,
validator: HttpCheck
): ScenarioBuilder =
scenario(scn).during(config.duration) {
exec(session => {
val daysFromNow = nextInt(180) + 1 //avoid daysFromNow = 0 because current date is invalid for some locale half a world away
val siteAndLang = chooseFrom(siteLangs)
session
.set("siteId", siteAndLang.substring(0, siteAndLang.indexOf("/")))
.set("locale", siteAndLang.substring(siteAndLang.indexOf("/") + 1))
.set("htlId", hotelIds)
.set("startDate", computeDateString(daysFromNow))
.set(
"endDate",
computeDateString(
daysFromNow + lengthOfStay(nextInt(lengthOfStay.length))
)
)
.set("regionId", regionIds)
.set("adults", chooseFrom(adultCounts))
}).exec(addCookie(Cookie("DUAID", DUAID)))
.exec(
http(scn)
.get(s"$endpoint/$path")
.header("Trace-ID", session => getTraceID())
.signatureCalculator(new CommonSignatureCalculator(DUAID))
.check(status is 200)
.check(validator)
)
}
private def validator = (jsonPath("$.hotelId").exists)
buildScenario(
"a_get_end_point",
"some scenario",
validator
)
