Forecasting values across categories using Prophet and Python

See also questions close to this topic

• Poor Accuracy of Gradient Descent Perceptron

  I'm trying to make a start with neural networks from the very beginning. This means starting off toying with perceptrons. At the moment I'm trying to implement batch gradient descent. The guide I'm following provided the following pseudocode:
  

  I've tried implementing as below with some dummy data and noticed it isn't particularly accurate. It converges, to what I presume is some local minima.

  My question is:
  What ways are there for me to check that this is infact a local minima, I've been looking into how to plot this, but I'm unsure how to actually go about doing this. In addition to this, is there a way to achieve a more accurate result using gradient descent? Or would I have to use a more complex approach, or possibly run it numerous times starting from different random weights to try and find the global minimum?

  I had a look about the forum before posting this, but didn't find much information that made me feel confident what I'm doing here or what's happening is in fact correct, so any helps would be great.

  import pandas as pd
  import numpy as np
  import random
  import math
  
  
  def main():
  
      learningRate = 0.1
      np.random.seed(1)
  
      trainingInput = np.asmatrix([
                [1, -1],
                [2, 1],
                [1.5, 0.5],
                [2, -1],
                [1, 2]
              ])
  
      biasAccount = np.ones((5,1))
      trainingInput = np.append(biasAccount, trainingInput, axis=1)
      trainingOutput = np.asmatrix([
                  [0],
                  [1],
                  [0],
                  [0],
                  [1]
              ])
  
  
  
      weights = 1 * np.random.random((3,1))-1
  
      for iteration in range(10000):
          prediction = np.dot(trainingInput, weights)
  
          print("Weights: \n" + str(weights))
  
          print("Prediction: \n" + str(prediction))
  
          error = trainingOutput - prediction
  
          print("Error: \n" + str(error))
  
          intermediateResult = np.dot(error.T, trainingInput)
          delta = np.dot(learningRate, intermediateResult)
  
          print("Delta: \n" + str(delta))
  
          weights += delta.T
  
  
  main()
  
• Difficulty with recursion in Python

  Here is a part of my code:

      estado_atual=0 
      lista_termino=[]
      for estado_inicial in lista_estados_iniciais:
          estado_atual=estado_inicial
          for i in range(len(palavra)):
              simbolo=palavra[i]
              if simbolo in dic_transicoes[estado_atual]:
                  for j in range(len(dic_transicoes[estado_atual][simbolo])):
                      estado_atual=dic_transicoes[estado_atual][simbolo][j]
              else:
                  estado_atual=0              
          lista_termino.append(estado_atual)
  

  The variable "dic_transicoes" is defined as

      dic_transicoes = defaultdict(lambda: defaultdict( list))
  

  and your content is:

      defaultdict(<function <lambda> at 0x7fbfc5eb1488>,
          {'1': defaultdict(<class 'list'>, {'1': ['2']}),
           '2': defaultdict(<class 'list'>, {'1': ['2', '3'], '2': ['2']})})
  

  The variable "palavra" is 112 (for example)

  I'm having trouble putting this part of code recursive:

          for j in range(len(dic_transicoes[estado_atual][simbolo])):
                  estado_atual=dic_transicoes[estado_atual][simbolo][j]
  

  My goal is to make transitions that represent a graph. The dictionary represents a transition from a to b by the symbol c for example, and in the case of the code this is represented by

       dic_transicoes[estado_atual][simbolo][j]
  

  What I want is that when I have more than one transition through a symbol, as is the case of the transition from 2 to 2 and 2 to 3 with the symbol 1, I can check the next state as if it were in parallel, and to this I thought of using recursion.

  Any ideia how can I do this?
• Stream Stream Data to Text or CSV format

  I want to read in the values of AINO - AIN11 and output them into different rows so the 1st row would read in the scans AINO - AIN11 then move to the next row and read in the values again and so on until the end of the loop (6000 scans currently). Right now it just separates each value and doesn't discern from scan 1 or 2 etc. I want to separate them out to differentiate between AINO-AIN1, AIN2-AIN3 and so forth. Once I separate them out I want to create 6 columns which produce the end value of AINO-AIN1, AIN2-AIN3, AIN4-AIN5, AIN6-AIN7, AIN8-AIN9, AIN10-AIN11. I'm rather new to coding/python and we're using labjack to read in the values to the file. Any help or idea would be appreciated.

  from datetime import datetime
  import sys
  
  from labjack import ljm
  
  file = open ("LabjackT7.txt", "w")
  
  
  MAX_REQUESTS = 1  # The number of eStreamRead calls that will be performed.
  
  # Open first found LabJack
  handle = ljm.openS("ANY", "ANY", "ANY")  # Any device, Any connection, Any identifier
  #handle = ljm.openS("T7", "ANY", "ANY")  # T7 device, Any connection, Any identifier
  #handle = ljm.openS("T4", "ANY", "ANY")  # T4 device, Any connection, Any identifier
  #handle = ljm.open(ljm.constants.dtANY, ljm.constants.ctANY, "ANY")  # Any device, Any connection, Any identifier
  
  info = ljm.getHandleInfo(handle)
  print("Opened a LabJack with Device type: %i, Connection type: %i,\n"
        "Serial number: %i, IP address: %s, Port: %i,\nMax bytes per MB: %i" %
        (info[0], info[1], info[2], ljm.numberToIP(info[3]), info[4], info[5]))
  
  deviceType = info[0]
  
  # Stream Configuration
  aScanListNames = ["AIN0", "AIN1", "AIN2", "AIN3", "AIN4", "AIN5", "AIN6", "AIN7", "AIN8", "AIN9", "AIN10", "AIN11"]  # Scan list names to stream
  numAddresses = len(aScanListNames)
  aScanList = ljm.namesToAddresses(numAddresses, aScanListNames)[0]
  scanRate = 10000
  scansPerRead = int(scanRate / 2)
  
  try:
      # When streaming, negative channels and ranges can be configured for
      # individual analog inputs, but the stream has only one settling time and
      # resolution.
  
      if deviceType == ljm.constants.dtT4:
          # LabJack T4 configuration
  
          # AIN0 and AIN1 ranges are +/-10 V, stream settling is 0 (default) and
          # stream resolution index is 0 (default).
          aNames = ["AIN0_RANGE", "AIN1_RANGE", "STREAM_SETTLING_US",
                    "STREAM_RESOLUTION_INDEX"]
          aValues = [10.0, 10.0, 0, 0]
      else:
          # LabJack T7 and other devices configuration
  
          # Ensure triggered stream is disabled.
          ljm.eWriteName(handle, "STREAM_TRIGGER_INDEX", 0)
  
          # Enabling internally-clocked stream.
          ljm.eWriteName(handle, "STREAM_CLOCK_SOURCE", 0)
  
          # All negative channels are single-ended, AIN0 and AIN1 ranges are
          # +/-10 V, stream settling is 0 (default) and stream resolution index
          # is 0 (default).
          aNames = ["AIN_ALL_NEGATIVE_CH", "AIN0_RANGE", "AIN1_RANGE",
                    "STREAM_SETTLING_US", "STREAM_RESOLUTION_INDEX"]
          aValues = [ljm.constants.GND, 10.0, 10.0, 0, 0]
      # Write the analog inputs' negative channels (when applicable), ranges,
      # stream settling time and stream resolution configuration.
      numFrames = len(aNames)
      ljm.eWriteNames(handle, numFrames, aNames, aValues)
  
      # Configure and start stream
      scanRate = ljm.eStreamStart(handle, scansPerRead, numAddresses, aScanList, scanRate)
      print("\nStream started with a scan rate of %0.0f Hz." % scanRate)
  
      print("\nPerforming %i stream reads." % MAX_REQUESTS)
      start = datetime.now()
      totScans = 0
      totSkip = 0  # Total skipped samples
  
  
  
      i = 1
      while i <= MAX_REQUESTS:
          ret = ljm.eStreamRead(handle)
  
          aData = ret[0]
          scans = len(aData) / numAddresses
          totScans += scans
  
          # Count the skipped samples which are indicated by -9999 values. Missed
          # samples occur after a device's stream buffer overflows and are
          # reported after auto-recover mode ends.
          curSkip = aData.count(-9999.0)
          totSkip += curSkip
  
          print("\neStreamRead %i" % i)
          ainStr = ""
          for j in range(0, numAddresses):
              ainStr += "%s = %0.5f, " % (aScanListNames[j], aData[j])
          print("  1st scan out of %i: %s" % (scans, ainStr))
          print("  Scans Skipped = %0.0f, Scan Backlogs: Device = %i, LJM = "
                "%i" % (curSkip/numAddresses, ret[1], ret[2]))
  
          i += 1
  #Data Stream string of last eStreamRead
      file.write('%s'% aData)
      file.close()
      end = datetime.now()
  
      print("\nTotal scans = %i" % (totScans))
      tt = (end - start).seconds + float((end - start).microseconds) / 1000000
      print("Time taken = %f seconds" % (tt))
      print("LJM Scan Rate = %f scans/second" % (scanRate))
      print("Timed Scan Rate = %f scans/second" % (totScans / tt))
      print("Timed Sample Rate = %f samples/second" % (totScans * numAddresses / tt))
      print("Skipped scans = %0.0f" % (totSkip / numAddresses))
  except ljm.LJMError:
      ljme = sys.exc_info()[1]
      print(ljme)
  except Exception:
      e = sys.exc_info()[1]
      print(e)
  
  try:
      print("\nStop Stream")
      ljm.eStreamStop(handle)
  except ljm.LJMError:
      ljme = sys.exc_info()[1]
      print(ljme)
  except Exception:
      e = sys.exc_info()[1]
      print(e)
  
  # Close handle
  ljm.close(handle)
  
• how to write only header with spark

  My situation:

  I have a set of sources and I have to pass them through layers of data, suppose that I have the layers A, B and C. Sometimes, any source lands in A layer with no data, only the header of the source, in my case, all data in A is avro. Then I have to pass it from A to B, in my case, layer B can be csv. Recently, the requires of layer B change and now I have parquet files too. I need the files because layer C need something to read, the header at least.

  My problem:

  It is when I have to parse that only header avro file to parquet file. Is there any solution using Spark/scala that can write only the header of a avro, parquet, etc format files?

  I have a code that can parse only headers to csv, just listing the columns and writing that as csv or plainText but when I try to write in avro or parquet, it only writes the _SUCCESS flag of spark. I have used the different save modes and properties that I have found and spark accepts.

  For more information, I use spark 2.3.1 version, scala 2.11.11
• Thread waiting for sun.misc.Unsafe.park in spark createDataFrame

  I am using spark to filter out data from a JavaRDD dataframe. It has around 2900000 record of which I need to filter 29000. There is no memory issue as I am using a super server with enought memory.

  Below is the code

      JavaRDD<SomeClass> sample = dataframe.javaRDD().map{..........};
      Dataset<Row> data= spark.createDataFrame(sample.rdd(), SomeClass.class).filter(col("column1").isNotNull());
  

  It just hangs on 2nd line for hours. When I tried checking the debug point by pausing the debug I was able to find the code was stuck at sun.misc.Unsafe.park.

  I checked similar answers none helped.

  There is no CPU utilization as well. And there can't be any memory issue as I am using super server.
• Sum of multiple columns based on condition

  I want to sum up an array of columns per row based on a condition. I found a way to do it but it does not seem like a good idea when I have 20+ columns to sum up since it would generate an additional column for each one.

  Wanted result is: Sum of values for all columns that end with "_val" where value is either 0 or 1 (or <2, i just want to exclude the value 3 right now)

   val df1 = Seq(
    ("id1", 1, 0, 3),
    ("id2", 0, 0, 3),
    ("id3", 1, 1, 3))
    .toDF("id", "bla_val", "blub_val", "bli_val")
  

  My solution with the wanted result in the column sum

  val channelNames = df1.schema.fieldNames.filter(_.endsWith("_val"))
  val ch = channelNames.map(x => col(x+"_redval"))
  
  val df2 = df1.select(col("*") +: (channelNames.map(c =>
    when(col(c) === 1, lit(1))
      .otherwise(lit(0)).as(c+"_redval"))): _*) 
  
  val df3 = df2.withColumn("sum", ch.reduce(_+_))
  df3.show()
  

  Example output:

  +---+-------+--------+-------+--------------+---------------+--------------+---+ | id|bla_val|blub_val|bli_val|bla_val_redval|blub_val_redval|bli_val_redval|sum| +---+-------+--------+-------+--------------+---------------+--------------+---+ |id1| 1| 0| 3| 1| 0| 0| 1| |id2| 0| 0| 3| 0| 0| 0| 0| |id3| 1| 1| 3| 1| 1| 0| 2| +---+-------+--------+-------+--------------+---------------+--------------+---+
• Why the following code in pyspark giving precission problem?

  df1.select("shp_nbr","shp_disc_amt","shp_fgt_chrg_amt").where(col("shp_nbr")=="CH7812705893").withColumn('shp_disc_per_overall',round((when(col("shp_disc_amt").isNull(),0).otherwise(col("shp_disc_amt"))/when(col("shp_fgt_chrg_amt")==0,1).when(col("shp_fgt_chrg_amt").isNull(),1).otherwise(col("shp_fgt_chrg_amt")))*100,2)).show()

  result: |CH7812705893| 21.61| 26.30| 82.00|

  df1.select("shp_nbr","shp_disc_amt","shp_fgt_chrg_amt").where(col("shp_nbr")=="CH7812705893").withColumn('EDW_DISCOUNT_PERCENT',round((when(col("shp_disc_amt").isNull(),0).otherwise(col("shp_disc_amt"))/col("shp_fgt_chrg_amt"))*100,2)).show()

  result: CH7812705893| 21.61| 26.30| 82.17
• Check for empty row within spark dataframe?

  Running over several csv files and i am trying to run and do some checks and for some reason for one file i am getting a NullPointerException and i am suspecting that there are some empty row.

  So i am running the following and for some reason it gives me an OK output:

  check_empty = lambda row : not any([False if k is None else True for k in row])
  check_empty_udf = sf.udf(check_empty, BooleanType())
  df.filter(check_empty_udf(sf.struct([col for col in df.columns]))).show()
  

  I am missing something within the filter function or we can't extract empty rows from dataframes.
• PyPark error in Spark's mapping since ElasticSearch's mapping

  in database's elasticSearch I have this mapping:

  {
  "parcel":{
      "mappings":{
          "_doc":{
              "properties":{
              "id":{
                  "type":"text"
              },
              "location":{
                  "type":"geo_shape"
              },
              "properties":{
                  "properties":{
                      "adjacentParcels":{
                          "properties":{
                          "contactLength":{
                              "type":"float"
                          },
                          "id":{
                              "type":"text"
                          }
                          }
                      },
                      "area":{
                          "type":"float"
                      },
                      "commune":{
                          "type":"text"
                      },
                      "constructibilities":{
                          "type":"nested",
                          "properties":{
                          "construbilitiesByBasementLevels":{
                              "type":"float"
                          },
                          "constructibilityMax":{
                              "type":"float"
                          },
                          "constructibilityWithProspects":{
                              "type":"float"
                          },
                          "type":{
                              "type":"text"
                          }
                          }
                      },
                      "contenance":{
                          "type":"long"
                      },
                      "created":{
                          "type":"date"
                      },
                      "id":{
                          "type":"text"
                      },
                      "Price":{
                          "type":"long"
                      },
                      "zonings":{
                          "properties":{
                          "etiquette":{
                              "type":"text"
                          },
                          "ratio":{
                              "type":"float"
                          },
                          "rules":{
                              "type":"nested",
                              "properties":{
                                  "emprise":{
                                      "type":"text"
                                  },
                                  "espaceVert":{
                                      "type":"text"
                                  },
                                  "hauteur":{
                                      "type":"text"
                                  },
                                  "type":{
                                      "type":"text"
                                  }
                              }
                          }
                          }
                      }
                  }
              },
              "type":{
                  "type":"text"
              }
              }
          }
      }
  }
  }
  

  and I do this mapping on pySpark:

  arrayRules= StructType([
      StructField("emprise", StringType()),
      StructField("espaceVert", StringType()),
      StructField("hauteur", StringType()),
      StructField("type", StringType()),
  ])
  
  arrayLocation= StructType([
      StructField("coordinates", ArrayType(FloatType())),
      StructField("type", StringType()),
  ])
  
  arrayZonings= StructType([
      StructField("etiquette", StringType()),
      StructField("ratio", FloatType()),
      StructField("rules", ArrayType(arrayRules)),
  ])
  
  arrayConstructibilities= StructType([
      StructField("construbilitiesByBasementLevels", ArrayType(FloatType())),
      StructField("constructibilityMax", FloatType()),
      StructField("constructibilityWithProspects", FloatType()),
      StructField("type", StringType()),
  ])
  
  arrayAdjacentParcels= StructType([
      StructField("contactLength", FloatType()),
      StructField("id", StringType()),
  ])
  
  stProperties= StructType([
      StructField("adjacentParcels", ArrayType(arrayAdjacentParcels)),
      StructField("area", FloatType()),
      StructField("constructibilities", ArrayType(arrayConstructibilities)),
      StructField("created", DateType()),
      StructField("id", StringType()),
      StructField("price", IntegerType()),
      StructField("zonings", ArrayType(arrayZonings)),
  ])
  
  schema= StructType([
      StructField("id", StringType()),
      StructField("properties", stProperties),
      StructField("type", StringType()),
      StructField("location", ArrayType(arrayLocation)),
  ])
  
  parcel= sqlContext.read.format("org.elasticsearch.spark.sql").schema(schema).load("parcel")
  parcel.printSchema()
  

  I get this schema:

  |-- id: string (nullable = true)
  |-- properties: struct (nullable = true)
  |    |-- adjacentParcels: array (nullable = true)
  |    |    |-- element: struct (containsNull = true)
  |    |    |    |-- contactLength: float (nullable = true)
  |    |    |    |-- id: string (nullable = true)
  |    |-- area: float (nullable = true)
  |    |-- constructibilities: array (nullable = true)
  |    |    |-- element: struct (containsNull = true)
  |    |    |    |-- construbilitiesByBasementLevels: array (nullable = true)
  |    |    |    |    |-- element: float (containsNull = true)
  |    |    |    |-- constructibilityMax: float (nullable = true)
  |    |    |    |-- constructibilityWithProspects: float (nullable = true)
  |    |    |    |-- type: string (nullable = true)
  |    |-- created: date (nullable = true)
  |    |-- id: string (nullable = true)
  |    |-- price: integer (nullable = true)
  |    |-- zonings: array (nullable = true)
  |    |    |-- element: struct (containsNull = true)
  |    |    |    |-- etiquette: string (nullable = true)
  |    |    |    |-- ratio: float (nullable = true)
  |    |    |    |-- rules: array (nullable = true)
  |    |    |    |    |-- element: struct (containsNull = true)
  |    |    |    |    |    |-- emprise: string (nullable = true)
  |    |    |    |    |    |-- espaceVert: string (nullable = true)
  |    |    |    |    |    |-- hauteur: string (nullable = true)
  |    |    |    |    |    |-- type: string (nullable = true)
  |-- type: string (nullable = true)
  |-- location: array (nullable = true)
  |    |-- element: struct (containsNull = true)
  |    |    |-- coordinates: array (nullable = true)
  |    |    |    |-- element: float (containsNull = true)
  |    |    |-- type: string (nullable = true)
  

  but there is for each table this element which must cause an error:

  -- element: struct (containsNull = true)
  

  and I get this error:

  Caused by: org.elasticsearch.hadoop.EsHadoopIllegalStateException: Field 'properties.constructibilities.constructibilityMax' not found; typically this occurs with arrays which are not mapped as single value
  at org.elasticsearch.spark.sql.RowValueReader$class.rowColumns(RowValueReader.scala:51)
  at org.elasticsearch.spark.sql.ScalaRowValueReader.rowColumns(ScalaEsRowValueReader.scala:32)
  at org.elasticsearch.spark.sql.ScalaRowValueReader.createMap(ScalaEsRowValueReader.scala:69)
  at org.elasticsearch.hadoop.serialization.ScrollReader.map(ScrollReader.java:968)
  at org.elasticsearch.hadoop.serialization.ScrollReader.readListItem(ScrollReader.java:875)
  at org.elasticsearch.hadoop.serialization.ScrollReader.list(ScrollReader.java:927)
  at org.elasticsearch.hadoop.serialization.ScrollReader.read(ScrollReader.java:833)
  at org.elasticsearch.hadoop.serialization.ScrollReader.map(ScrollReader.java:1004)
  at org.elasticsearch.hadoop.serialization.ScrollReader.read(ScrollReader.java:846)
  at org.elasticsearch.hadoop.serialization.ScrollReader.map(ScrollReader.java:1004)
  at org.elasticsearch.hadoop.serialization.ScrollReader.read(ScrollReader.java:846)
  at org.elasticsearch.hadoop.serialization.ScrollReader.readHitAsMap(ScrollReader.java:602)
  at org.elasticsearch.hadoop.serialization.ScrollReader.readHit(ScrollReader.java:426)
  ... 27 more
  

  would you be able to explain me how to make ArrayType because I did not find a tutorial with complex schema.

  thank you so much.
• Unsupported literal type class scala.runtime.BoxedUnit

  I am trying to filter a column of a dataframe read from oracle as below

  import org.apache.spark.sql.functions.{col, lit, when}
  
  val df0  =  df_org.filter(col("fiscal_year").isNotNull())
  

  When I do it I am getting below error:

  java.lang.RuntimeException: Unsupported literal type class scala.runtime.BoxedUnit ()
  at org.apache.spark.sql.catalyst.expressions.Literal$.apply(literals.scala:77)
  at org.apache.spark.sql.catalyst.expressions.Literal$$anonfun$create$2.apply(literals.scala:163)
  at org.apache.spark.sql.catalyst.expressions.Literal$$anonfun$create$2.apply(literals.scala:163)
  at scala.util.Try.getOrElse(Try.scala:79)
  at org.apache.spark.sql.catalyst.expressions.Literal$.create(literals.scala:162)
  at org.apache.spark.sql.functions$.typedLit(functions.scala:113)
  at org.apache.spark.sql.functions$.lit(functions.scala:96)
  at org.apache.spark.sql.Column.apply(Column.scala:212)
  at com.snp.processors.BenchmarkModelValsProcessor2.process(BenchmarkModelValsProcessor2.scala:80)
  at com.snp.utils.Utils$$anonfun$getAllDefinedProcessors$1.apply(Utils.scala:30)
  at com.snp.utils.Utils$$anonfun$getAllDefinedProcessors$1.apply(Utils.scala:30)
  at com.sp.MigrationDriver$$anonfun$main$6$$anonfun$apply$2.apply(MigrationDriver.scala:140)
  at com.sp.MigrationDriver$$anonfun$main$6$$anonfun$apply$2.apply(MigrationDriver.scala:140)
  at scala.Option.map(Option.scala:146)
  at com.sp.MigrationDriver$$anonfun$main$6.apply(MigrationDriver.scala:138)
  at com.sp.MigrationDriver$$anonfun$main$6.apply(MigrationDriver.scala:135)
  at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:733)
  at scala.collection.Iterator$class.foreach(Iterator.scala:891)
  at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
  at scala.collection.MapLike$DefaultKeySet.foreach(MapLike.scala:174)
  at scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:732)
  at com.sp.MigrationDriver$.main(MigrationDriver.scala:135)
  at com.sp.MigrationDriver.main(MigrationDriver.scala)
  

  Any idea what am I doing wrong here and how to fix this?
• Is it possible to build spark code on fly and execute?

  I am trying to create a generic function to read a csv file using databriks CSV READER.But the option's are not mandatory it can differ based on the my input json configuration file.

  Example1 :

   "ReaderOption":{
  
                      "delimiter":";",
                      "header":"true",
                      "inferSchema":"true",
                      "schema":"""some custome schema.."""
              },
  

  Example2:

     "ReaderOption":{
  
                          "delimiter":";",
                          "schema":"""some custome schema.."""
                  },
  

  Is it possible to construct options or the entire read statement in runtime and run in spark ?

  like below,

    def readCsvWithOptions(): DataFrame=
  {
  
    val options:Map[String,String]= Map("inferSchema"->"true")
    val readDF = jobContext.spark.read.format("com.databricks.spark.csv")
     .option(options)
    .load(inputPath)
    readDF
  }
  
• Pyspark .partitionBy().parquet() doesn't save files

  I am currently working inAzure Databricks. I have one consolidated spark dataframe with such columns as 'year', 'month' and 'day'.

  What I need to do is to partition it based on these columns and save all of them inside my azure data lake store( in the parquet format ). So for each partition the path should look like "adl/Data/UserData/year/month/day/"

  From my understanding, it should be simple as that:

   new_user_dataframe.write.partitionBy("year", "month", "day").parquet("adl/Data/UserData")
  

  The spark job is running in my notebook but the parent folder remains empty. It works fine without partitionByclause though. So I can write new_user_dataframe.write.parquet(full path for some specific day)but it is not what I need.

  I was thinking about doing something like new_user_dataframe.write.partitionBy("year", "month", "day").parquet("adl/Data/UserData/y=" + year + "/m=" + month + "/d=" + day). However, I don't know how to access these variables in order to pass them into the path.

  Thank you for any advice
• In a dataset for forecasting with Prophet, the date column should be ordered?

  I noticed problems of convergence for the Prophet algorithm when your date column, the ds column, is not properly ordered. How could this be explained?
• 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: