Hello guys , i am new to spark and scala. i have two dataframes i want to join them in order to get a value based on where a date is situtaed among a set of start and end date .Here is what i did as example.Thanks i advance Here is an example val histocaisse = spark.read.option("header", true).csv("C:\\Users\\MHT\\Desktop\\histocaisse.csv") .withColumn("pos_id",'pos_id.cast(LongType)) .withColumn("DEBUT",'DEBUT.cast(DateType)) .withColumn("FIN",'FIN.cast(DateType)) .withColumn("PRIXV",'PRIXV.cast(DoubleType)) val test = spark.read.option("header", true).csv("C:\\Users\\MHT\\Desktop\\test.csv") .withColumn("synch_date",'synch_date.cast(DateType)) test.show val bis = spark.read.option("header", true).csv("C:\\Users\\MHT\\Desktop\\bis.csv") .withColumn("synch_date",'synch_date.cast(DateType)) .select("pos_id","synch_date","sale_price") +------+----------+----------+-----+ |pos_id| DEBUT| FIN|PRIXV| +------+----------+----------+-----+ | 1|2000-01-02|2000-01-07| 2.5| | 2|2000-02-18|2000-02-25| 2.3| | 3|2000-03-02|2000-03-25| 14.0| +------+----------+----------+-----+ +------+----------+ |pos_id|synch_date| +------+----------+ | 1|2000-01-03| | 2|2000-02-19| | 3|2000-03-20| +------+----------+ +------+----------+----------+ |pos_id|synch_date|sale_price| +------+----------+----------+ | 1|2000-01-03| 2.5| | 2|2000-02-19| 2.3| | 3|2000-03-20| 14| +------+----------+----------+ ... View more

I am applying correlation on a csv file using apache spark, when loading data i am obliged to skipe the first row as a header which are columns in the dataset otherwise i can't load the data. i get the correlation computed but when i got the correlation matrix, i can't add the columns name as a header in the new matrix please would you help me get the matrix with its header ,thanks ,this what i did import org.apache.spark.mllib.linalg.{Vector,Vectors} import org.apache.spark.mllib.stat.Statistics import org.apache.spark.mllib.linalg.Matrix import org.apache.spark.rdd.RDD val data = sc.textFile(strfilePath).mapPartitionsWithIndex {case(index, iterator)=>if(index ==0) iterator.drop(1)else iterator } val inputMatrix = data.map { line =>val values = line.split(",").map(_.toDouble)Vectors.dense(values)} val correlationMatrix=Statistics.corr(inputMatrix,"pearson") ... View more

i am trying to load data from mongodb to spark as a dataframe , here is what i did but i got an error you will find below any help will be appreciated import org.apache.spark.sql.SparkSession import com.mongodb.spark.sql._ import com.mongodb.spark._ import org.bson.Document import com.mongodb.spark.config._ object MongoDB extends App{ try {val sparkSession =SparkSession.builder().master("local").getOrCreate() def makeMongoURI(uri:String,database:String,collection:String)=(s"${uri}/${database}.${collection}") val mongoURI ="mongodb://127.0.0.1:27017" val Conf= makeMongoURI(mongoURI,"io","thing") val readConfigintegra:ReadConfig=ReadConfig(Map("uri"->Conf)) val df3 = sparkSession.sqlContext.loadFromMongoDB(ReadConfig(Map("uri"->"mongodb://127.0.0.1:27017/io.thing"))) df3.printSchema()} catch{ case t:Throwable=> t.printStackTrace()// TODO: handle error println(t.getMessage) } } ... View more

Hello guys i have the following code i did , i want to get a new dataframe from a dataframe in which each value=176-old value any help please val spark = SparkSession.builder.master("local").appName("my-spark-app").getOrCreate val df=spark.read.option("header",true).csv("hdfs://localhost:9000/data1/data_with_missing_values.csv").cache() val dfNumber= df.describe(df.columns:_*).filter(col("summary") === "count").cache() dfNumber.show println("Number of rows :" + df.count()) ... View more

hi guys i want to apply a backtit to all column existing in a select request with `id -u` a backtickhow to would be my code in order to select all column in the aggs because all my column 'name include dot val spark = SparkSession.builder.master("local").appName("my-spark-app").getOrCreate() val df = spark.read.option("inferSchema", "true").option("header", "true").csv("D:\\ProcessDataSet\\anis_data\\Set _1Mud Pumps_Merged.csv") val aggs = df.columns.map(c => stddev(c).as(c)) val stddevs = df.select(aggs: _*)// what should i do to aggs to call with a `column_name` stddevs.show(5,false) ... View more

hi guys am using dataset with column name containing "." , here is the code am using and after that the issue i got , i think there is a problem with the "." character ,what should i do ? Thanks. val spark = SparkSession.builder.master("local").appName("my-spark-app").getOrCreate() val df = spark.read.option("inferSchema", "true").option("header", "true").csv("D:\ProcessDataSet\anis_data\Set _1Mud Pumps_Merged.csv") val aggs = df.columns.map(c => stddev(c).as(c)) val stddevs = df.select(aggs: _*) stddevs.show(5,false) ... View more

i have 4 csv files , i want to join and merge these files into one files based on a column timestamps to get one file.using spark or hadoop Please any help would be appreciated ... View more

Hi guys i am trying to save a dataframe to a csv file , that contains a timestamp. The problem that this column changes of format one written in the csv file .when showing via df.show i got a correct format when i check the csv file i got this format i also tried some think like this ,and still got the same problem finalresult.coalesce(1).write.option("header",true).option("inferSchema","true").option("dateFormat","yyyy-MM-dd HH:mm:ss").csv("C:/mydata.csv") val spark =SparkSession.builder.master("local").appName("my-spark-app").getOrCreate()val df = spark.read.option("header",true).option("inferSchema","true").csv("C:/Users/mhattabi/Desktop/dataTest2.csv")//val df = spark.read.option("header",true).option("inferSchema", "true").csv("C:\dataSet.csv\datasetTest.csv")//convert all column to numeric value in order to apply aggregation function df.columns.map { c =>df.withColumn(c, col(c).cast("int"))}//add a new column inluding the new timestamp columnval result2=df.withColumn("new_time",((unix_timestamp(col("time"))/300).cast("long")*300).cast("timestamp")).drop("time")val finalresult=result2.groupBy("new_time").agg(result2.drop("new_time").columns.map((_ ->"mean")).toMap).sort("new_time")//agg(avg(all columns..) finalresult.coalesce(1).write.option("header",true).option("inferSchema","true").csv("C:/mydata.csv") ... View more

I have dobe this code ,my question is for the function cast data type ,how can i cast all columns'datatype included in dataset at the same time except the column timestamp, and the other question is how to apply function avg on all column except also column timestamp,thanks a lot . val df = spark.read.option("header",true).option("inferSchema", "true").csv("C:/Users/mhattabi/Desktop/dataTest.csv") val result=df.withColumn("new_time",((unix_timestamp(col("time")) /300).cast("long") * 300).cast("timestamp")) result("value").cast("float")//here the first question val finalresult=result.groupBy("new_time").agg(avg("value")).sort("new_time")//here the second question about avg finalresult.coalesce(1).write.format("com.databricks.spark.csv").option("header", "true").save("C:/mydata.csv") ... View more

Hi guys Please , i need your help please , i need to aggregate a dataset based on a 5 minute interval and aggregating based on average function ,here you may find input and expected output .,your help will be highly appreciated ,the first column is a timestamp column and am using scala language ... View more

Hi guys i am working with scala and spark , i would like what is the best api or the best way to aggregate data based on interval of timestamps ,knowing that my data includes column timestamp that are taken every second , i would like to get a new dataframe based on aggregation and interval .Example my timestamp column is taken every second , i would like to know how would be my data if the timestamp was taken every minute ? val spark = SparkSession.builder.master("local").appName("my-spark-app").getOrCreate() val df = spark.read.option("header",true).csv("C:/Users/mhattabi/Desktop/Clean _data/Mud_Pumps _Cleaned/Set_1_Mud_Pumps_Merged.csv") df("DateTime").cast("timestamp") Thanks any help would Any help would be appreciated ... View more

Hi guys i am trying this source code val spark = SparkSession.builder.master("local").appName("my-spark-app").getOrCreate() val df = spark.read.option("header",true).csv("C:/Users/mhattabi/Desktop/Clean _data/Mud_Pumps _Cleaned/Set_1_Mud_Pumps_Merged.csv") df("DateTime").cast("timestamp") df("ADCH_Mud Pumps.db.MudPump.2.On.value").cast("integer") val result=df.select( col("*"), date_format(df("DateTime"),"yyyy-MM-dd hh:mm").alias("DateTime")).groupBy(df("DateTime")) .agg(avg(df("ADCH_Mud Pumps.db.MudPump.2.On.value"))) result.show(5) It says what sould i do for the attribute it is really existing in my data set,Thanks ... View more

HI guys timestamp name count(b.name)------------------- ---- -------------2010-11-1610:32:22 John 22010-11-1610:35:12 John 72010-11-1610:36:34 John 12010-11-1610:37:45 John 22010-11-1610:48:26 John 82010-11-1610:55:00 John 92010-11-1610:58:08 John 2 this a current dataframe i have , what i want to do is to get a new dataframe having them grouped by 5 minute based on timestamp to get someting like this timestamp name count(b.name) ------------------- ---- ------------- 2010-11-16 10:30:00 John 2 2010-11-16 10:35:00 John 10 2010-11-16 10:40:00 John 0 2010-11-16 10:45:00 John 8 2010-11-16 10:50:00 John 0 2010-11-16 10:55:00 John 11 using spark and scala , Any idea thanks ... View more

hi guys i have a code written in pyspark any help to run it under scala thanks it is urgent please thanks from functools import reduce files =["/tmp/test_1.csv","/tmp/test_2.csv","/tmp/test_3.csv"] df = reduce(lambda x,y: x.unionAll(y), [sqlContext.read.format('com.databricks.spark.csv') .load(f, header="true", inferSchema="true") for f in files]) df.show() ... View more

hi guys Am new to spark ans scala,i have csv files that i want tomerge in the same csv file or dataframe i want just to handle them as if they are only one file Any help thanks ... View more

Hi I was able to load data remotely through the webhdfs rest api ,but it doesn't allow to load a big volume of data remotely.Is there any possibilty to load huge data remotely ,i ask if there is a hadoop api to do that ? Thank you ... View more

I am using HDP , i would like to use the webhdfs api of hadoop through .Net hadoop webhdfs api, i would like to access the hadoop in a remote machine through the webhdfs , i would like to know the URI that should be done to aceess hdp remotely using http.It should be something like http://host:port/webhdfs, what is the port to use Thank you ... View more

Hello guys I am running hadoop cluster using HDP , when i run hadoop there are two CLI of the datanode and the namenode , please is there is any possibilty to run them in background ? Thank you a lot ! ... View more

Hi @Sagar Shimpi I would like to know about the HDP Price just in case ,we want to use your product (HortonWorks Data Platform) to develop our personal software. Thank you ... View more