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 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

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