Forecasting values across categories using Prophet and Python

See also questions close to this topic

• Python Kafka Streaming API - Binning

  I am using python kafka stream binning example given in this, Python Kafka Streaming API

  I am able to generate the data using generator.py file given under winton-kafka-streams/examples/binning/, whereas when i run the binning.py file from the same folder, i got the below issue. Could someone help me, to resolve this?

  

  
• Change color of missing values in Seaborn heatmap

  Consider the example of missing values in the Seaborn documentation:

  corr = np.corrcoef(np.random.randn(10, 200))
  mask = np.zeros_like(corr)
  mask[np.triu_indices_from(mask)] = True
  sns.heatmap(corr, mask=mask, vmax=.3, square=True)
  

  

  How do I change the color of the missing values to, for example, black? The color of the missing values should be specified independent of the color scheme of the heatmap, it may not be present in the color scheme.

  I tried adding facecolor = 'black' but that didn't work. The color can be affected by e.g. sns.axes_style("white") but it isn't clear to me how that can be used to set an arbitrary color.
• Xpath + Scrapy + Python : data point couldn't be scraped

  This is the XML structure:

  <tr>
     <td>
         <font size="3">
             <strong>Location:</strong> 
             Hiranandani Gardens, Powai
         </font>
     </td> 
  </tr>  
  

  I want to extract : Hiranandani Gardens, Powai

  I tried with these:

  response.xpath('//font[@size="3"]/strong[text()="Location:"]/text()').extract()
  
  response.xpath('//font[@size="3"]/strong[text()="Location:"]').extract()
  

  Both returned an empty list.

  Note: we must have to use the text of tag, i.e., "Location:". Otherwise, there are many other places on the site where the same XML structure is used. So, it'll fetch many more unnecessary things apart from the desired value if the text of strong tag is not used.
• azure databricks writeStream is not working

  I am using Azure Databricks. When I tried to write steaming data after sentiment analysis into blob storage. I got error message job aborted. below is error message. can you please advise?

  import org.apache.spark.sql.streaming.Trigger.ProcessingTime

  val query =
    streamingDataFrame
      .writeStream
      .format("json")
      .option("path", "/mnt/blobimages/DatabricksSentimentPowerBI1")
      .option("checkpointLocation", "/mnt/blobimages/sample/check1")    
      .trigger(ProcessingTime("20 seconds"))
      .start()
  

  at org.apache.spark.sql.execution.datasources.FileFormatWriter$.write(FileFormatWriter.scala:224) at org.apache.spark.sql.execution.streaming.FileStreamSink.addBatch(FileStreamSink.scala:131) at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$runBatch$5$$anonfun$apply$17.apply(MicroBatchExecution.scala:549) at org.apache.spark.sql.execution.SQLExecution$$anonfun$withCustomExecutionEnv$1.apply(SQLExecution.scala:89) at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:175) at org.apache.spark.sql.execution.SQLExecution$.withCustomExecutionEnv(SQLExecution.scala:84) at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:126) at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$runBatch$5.apply(MicroBatchExecution.scala:547) at org.apache.spark.sql.execution.streaming.ProgressReporter$class.reportTimeTaken(ProgressReporter.scala:379) at org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:60) at org.apache.spark.sql.execution.streaming.MicroBatchExecution.org$apache$spark$sql$execution$streaming$MicroBatchExecution$$runBatch(MicroBatchExecution.scala:546) at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1$$anonfun$apply$mcZ$sp$1.apply$mcV$sp(MicroBatchExecution.scala:204) at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1$$anonfun$apply$mcZ$sp$1.apply(MicroBatchExecution.scala:172) at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1$$anonfun$apply$mcZ$sp$1.apply(MicroBatchExecution.scala:172) at org.apache.spark.sql.execution.streaming.ProgressReporter$class.reportTimeTaken(ProgressReporter.scala:379) at org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:60) at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1.apply$mcZ$sp(MicroBatchExecution.scala:172) at org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.execute(TriggerExecutor.scala:56) at org.apache.spark.sql.execution.streaming.MicroBatchExecution.runActivatedStream(MicroBatchExecution.scala:166) at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:293) at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:203) Caused by: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 0.0 failed 4 times, most recent failure: Lost task 0.3 in stage 0.0 (TID 4, 10.139.64.5, executor 0): java.lang.IllegalStateException: In partition 0 of demo_event_hub, with consumer group $Default, request seqNo 1339 is less than the received seqNo 1502. The earliest seqNo is 1501 and the last seqNo is 1537 at org.apache.spark.eventhubs.client.CachedEventHubsReceiver.checkCursor(CachedEventHubsReceiver.scala:115) at org.apache.spark.eventhubs.client.CachedEventHubsReceiver.org$apache$spark$eventhubs$client$CachedEventHubsReceiver$$receive(CachedEventHubsReceiver.scala:128) at org.apache.spark.eventhubs.client.CachedEventHubsReceiver$.receive(CachedEventHubsReceiver.scala:178) at org.apache.spark.eventhubs.rdd.EventHubsRDD.compute(EventHubsRDD.scala:122) at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:336) at org.apache.spark.rdd.RDD.iterator(RDD.scala:300) at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:42) at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:336) at org.apache.spark.rdd.RDD.iterator(RDD.scala:300) at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:42) at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:336) at org.apache.spark.rdd.RDD.iterator(RDD.scala:300) at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87) at org.apache.spark.scheduler.Task.run(Task.scala:112) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:384) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748)

  Driver stacktrace: at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1747) at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1735) at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1734) at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48) at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1734) at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:962) at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:962) at scala.Option.foreach(Option.scala:257) at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:962) at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1970) at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1918) at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1906) at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49) at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:759) at org.apache.spark.SparkContext.runJob(SparkContext.scala:2141) at org.apache.spark.sql.execution.datasources.FileFormatWriter$.write(FileFormatWriter.scala:194) ... 20 more Caused by: java.lang.IllegalStateException: In partition 0 of demo_event_hub, with consumer group $Default, request seqNo 1339 is less than the received seqNo 1502. The earliest seqNo is 1501 and the last seqNo is 1537 at org.apache.spark.eventhubs.client.CachedEventHubsReceiver.checkCursor(CachedEventHubsReceiver.scala:115) at org.apache.spark.eventhubs.client.CachedEventHubsReceiver.org$apache$spark$eventhubs$client$CachedEventHubsReceiver$$receive(CachedEventHubsReceiver.scala:128) at org.apache.spark.eventhubs.client.CachedEventHubsReceiver$.receive(CachedEventHubsReceiver.scala:178) at org.apache.spark.eventhubs.rdd.EventHubsRDD.compute(EventHubsRDD.scala:122) at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:336) at org.apache.spark.rdd.RDD.iterator(RDD.scala:300) at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:42) at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:336) at org.apache.spark.rdd.RDD.iterator(RDD.scala:300) at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:42) at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:336) at org.apache.spark.rdd.RDD.iterator(RDD.scala:300) at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87) at org.apache.spark.scheduler.Task.run(Task.scala:112) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:384) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748)
• Dataframe null values transformed to 0 after UDF. Why?

  How can Nulls be handled when accessing dataframe Row values? Does the Null pointer Exception really require to be handled manually? There must be a better solution.

  case class FirstThing(id:Int, thing:String, other:Option[Double])
  
  val df = Seq(FirstThing(1, "first", None), FirstThing(1, "second", Some(2)), FirstThing(1, "third", Some(3))).toDS
  df.show
  
  val list = df.groupBy("id").agg(collect_list(struct("thing", "other")).alias("mylist"))
  list.show(false)
  

  This fails with NPE:

  val xxxx = udf((t:Seq[Row])=> t.map(elem => elem.getDouble(1)))
  list.withColumn("aa", xxxx(col("mylist"))).show(false)
  

  This strangely gives 0:

  val xxxx = udf((t:Seq[Row])=> t.map(elem => elem.getAs[Double]("other")))
  list.withColumn("aa", xxxx(col("mylist"))).show(false)
  
  +---+-----------------------------------------+---------------+
  |id |mylist                                   |aa             |
  +---+-----------------------------------------+---------------+
  |1  |[[first,null], [second,2.0], [third,3.0]]|[0.0, 2.0, 3.0]|
  +---+-----------------------------------------+---------------+
  

  Sadly this approach which works fine with data frames/datasets fails as well:

  val xxxx = udf((t:Seq[Row])=> t.map(elem => elem.getAs[Option[Double]]("other")))
  list.withColumn("aa", xxxx(col("mylist"))).show(false)
  

  ClassCastException: java.lang.Double cannot be cast to scala.Option
• Validate Avro file using JSON schema

  We have requirement of validating avro file data using Json schema in Spark 2.0 JSON schema is of type draft 4.

  Requirement is validate every column from row with multiple rules like integer range, in between values, greater than a value etc. on same column and collect all errors. Can you please help on this.
• Spark (Pyspark) - Long delay between jobs

  I have a spark job that performs some data analysis and there is a very long time between each job starting (1-2 hours). I know that some work isn't parallelized and doesn't count as a job so doesn't show up in the resource manager UI. I was hoping to find out what is taking so long? Here is my function:

  def analyse(df, columns):
      return reduce(
          lambda a, b: a.union(b),
          (
              df.agg(
                  F.count(F.when(F.isnull(c), c)).alias('NULL Count'),
                  F.countDistinct(df[c]).alias('Distinct Values')
              ).select(F.lit(c).alias("Column Name"), "NULL Count", 'Distinct Values')
              for c in columns
          )
      )
  

  Any ideas?
• PySpark write CSV quote all non-numeric

  Is there a way to quote only non-numeric columns in the dataframe when output to CSV file using df.write.csv('path')?

  I know you can use the option quoteAll=True to quote all the columns but I only want to quote the string columns.

  I am using PySpark 2.2.0.
• Remove rows from dataframe based on condition in pyspark

  I have one dataframe with two columns:

  +--------+-----+
  |    col1| col2|
  +--------+-----+
  |22      | 12.2|
  |1       |  2.1|
  |5       | 52.1|
  |2       | 62.9|
  |77      | 33.3|
  

  I would like to create a new dataframe which will take only rows where

  "value of col1" > "value of col2"

  Just as a note the col1 has long type and col2 has double type

  the result should be like this:

  +--------+----+
  |    col1|col2|
  +--------+----+
  |22      |12.2|
  |77      |33.3|
  
• Azure Databricks vs ADLA for processing

  Presently, I have all my data files in Azure Data Lake Store. I need to process these files which are mostly in csv format. The processing would be running jobs on these files to extract various information for e.g.Data for certain periods of dates or certain events related to a scenario or adding data from multiple tables/files. These jobs run everyday through u-sql jobs in data factory(v1 or v2) and then sent to powerBI for visualization.

  Using ADLA for all this processing, I feel it takes a lot of time to process and seems very expensive. I got a suggestion that I should use Azure Databricks for the above processes. Could somebody help me with this direction in the difference between the two and if it would be helpful to shift? Can I modify all my U-sql jobs into the Databricks notebook format?
• Connect to Blob storage "no credentials found for them in the configuration"

  I'm working with Databricks notebook backed by spark cluster. Having trouble trying to connect to the Azure blob storage. I used this link and tried the section Access Azure Blob Storage Directly - Set up an account access key. I get no errors here:

  spark.conf.set(
    "fs.azure.account.key.<your-storage-account-name>.blob.core.windows.net",
    "<your-storage-account-access-key>")
  

  But receive errors when I try and do an 'ls' on the directory:

  dbutils.fs.ls("wasbs://<your-container-name>@<your-storage-account-name>.blob.core.windows.net/<your-directory-name>")
  
  shaded.databricks.org.apache.hadoop.fs.azure.AzureException: shaded.databricks.org.apache.hadoop.fs.azure.AzureException: Unable to access container <container name> in account <storage account name>core.windows.net using anonymous credentials, and no credentials found for them in the configuration.
  

  If there is a better way, please provide suggestion as well. thanks
• How to read a properties file in databricks in scala without using spark?

  I am able to read it as RDD

  val rdd  = sc.textFile("dbfs:/mnt/abc/XYZ.properties")
  

  but the file has very less data and i dont want to read it as rdd as it will split the data in different nodes

  so how can we read data without using spark ?
• Prophet Forecasting using R for multiple items

  I am very new to time series forecasting using Prophet in R. I am able to predict values for one single product using Prophet. Is there any way if i can use loop to generate forecast using Prophet for multiple products? The below code works absolutely fine for single product but i am trying to generate forecasts for multiple products

   library(prophet)
   df <- read.csv("Prophet.csv")
   df$Date<-as.Date(as.character(df$Date), format =  "%d-%m-%Y")
   colnames(df) <- c("ds", "y")
   m <- prophet(df)
   future <- make_future_dataframe(m, periods = 40)
   tail(future)
   forecast <- predict(m, future)
   write.csv(forecast[c('ds','yhat')],"Output_Prophet.csv")
   tail(forecast[c('ds', 'yhat', 'yhat_lower', 'yhat_upper')])
  

  Sample Dataset:

  
• R crashes when prophet model run in a loop

  R crashes when prophet model run in a loop and the termination of the loop is not in a particular fashion i.e. it crashes R randomly.

  for(j in 1:length(unique_code_train))
  {
  train1<- filter(train, CODE==unique_code_train[j])
  train1<-arrange(train1, BILLDATE)
  train1<-data.table(train1)
  names(train1)<-c("ds","y") # billdate= ds , sum(saleamt)= y
  m<-prophet(train1 ,changepoint.range = 1.0, seasonality.mode = 
  'multiplicative', holidays = holidays)
  future <- make_future_dataframe(m, periods = 26)
  forecast <- predict(m, future)
  forecast$yhat<-as.integer(forecast$yhat)
  }