title: How create parquet table in scala?
body: <p>I want to create a parquet table with certain types of fields:</p>
  <p>name_process: String
  id_session: Int
  time_write: LocalDate or Timestamp
  key: String
  value: String</p>
  <div class="s-table-container">
  <table class="s-table">
  <thead>
  <tr>
  <th>name_process</th>
  <th>id_session</th>
  <th>time_write</th>
  <th>key</th>
  <th>value</th>
  </tr>
  </thead>
  <tbody>
  <tr>
  <td>OtherClass</td>
  <td>jsdfsadfsf</td>
  <td>43434883477</td>
  <td>schema0.table0.csv</td>
  <td>Success</td>
  </tr>
  <tr>
  <td>OtherClass</td>
  <td>jksdfkjhka</td>
  <td>23212123323</td>
  <td>schema1.table1.csv</td>
  <td>Success</td>
  </tr>
  <tr>
  <td>OtherClass</td>
  <td>alskdfksjd</td>
  <td>23343212234</td>
  <td>schema2.table2.csv</td>
  <td>Failure</td>
  </tr>
  <tr>
  <td>ExternalClass</td>
  <td>sdfjkhsdfd</td>
  <td>34455453434</td>
  <td>schema3.table3.csv</td>
  <td>Success</td>
  </tr>
  </tbody>
  </table>
  </div>
  <p>I want to write such a table correctly. With the correct data types. Then I'm going to read the partitions from it. I'm trying to implement read and write. But it turns out badly so far.</p>
  <pre><code>def createHiveTable(implicit spark: SparkSession) {
  
    val schema = &quot;test_schema&quot;
    val table = &quot;test_table&quot;
    val partitionName = &quot;name_process&quot;
    val columnNames = &quot;name_process&quot; :: &quot;id_session&quot; :: &quot;time_write&quot; :: &quot;key&quot; :: &quot;value&quot; :: Nil
  
    spark.sql(s&quot;CREATE DATABASE IF NOT EXISTS test_db&quot;)
    //val createTableSql = s&quot;CREATE TABLE IF NOT EXISTS $schema.$table ($columnNames) PARTITIONED BY $partitionName STORED AS parquet&quot;
  
    val path = new File(&quot;.&quot;).getAbsolutePath ++ &quot;/src/test/data-lineage/test_data_journal.csv&quot;
  
    val df = spark.read.option(&quot;delimiter&quot;, &quot;,&quot;)
      .option(&quot;header&quot;, true)
      .csv(path)
  
    df.show()
  
    df.write.mode(SaveMode.Append).partitionBy(partitionName).format(&quot;parquet&quot;).saveAsTable(s&quot;test_db.$table&quot;)
  
  }
  </code></pre>
  <hr />
  <pre><code>def getLastSession(processName: String)(implicit spark: SparkSession): Unit = {
  
    val df = spark.read.parquet(s&quot;spark-warehouse/test_db.db/test_table/name_process=$processName&quot;)
                       .select(
                                col(&quot;name_process&quot;).cast(StringType),
                                col(&quot;id_session&quot;).cast(StringType),
                                col(&quot;time_write&quot;).cast(LongType),
                                col(&quot;key&quot;).cast(StringType),
                                col(&quot;value&quot;).cast(StringType)
                       )
  
    val lastTime = df.select(col(&quot;time_write&quot;)).select(max(&quot;time_write&quot;)).collect()(0).get(0)
    val lastSession = df.filter(col(&quot;time_write&quot;).equalTo(lastTime)).select(&quot;id_session&quot;).head().getString(0)
  
    println(lastSession)
    println(TimeStamp.getCurrentTime)
  }
  </code></pre>
  <p>logs from spark:</p>
  <pre><code>21/12/16 14:51:19 INFO TaskSchedulerImpl: Killing all running tasks in stage 3: Stage finished
  21/12/16 14:51:19 INFO DAGScheduler: Job 3 finished: parquet at DataLineageJournal.scala:28, took 0,076899 s
   
  org.apache.spark.sql.AnalysisException: cannot resolve '`name_process`' given input columns: [id_session, key, time_write, value];
  'Project [unresolvedalias(cast('name_process as string), None), cast(id_session#78 as string) AS id_session#86, cast(time_write#79 as bigint) AS time_write#87L, cast(key#80 as string) AS key#88, cast(value#81 as string) AS value#89]
  +- Relation[id_session#78,time_write#79,key#80,value#81] parquet
   
   
                  at org.apache.spark.sql.catalyst.analysis.package$AnalysisErrorAt.failAnalysis(package.scala:42)
                  at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$$nestedInanonfun$checkAnalysis$1$2.applyOrElse(CheckAnalysis.scala:155)
                  at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$$nestedInanonfun$checkAnalysis$1$2.applyOrElse(CheckAnalysis.scala:152)
                  at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformUp$2(TreeNode.scala:342)
                  at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:74)
                  at org.apache.spark.sql.catalyst.trees.TreeNode.transformUp(TreeNode.scala:342)
                  at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformUp$1(TreeNode.scala:339)
                  at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$mapChildren$1(TreeNode.scala:408)
                  at org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:244)
                  at org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:406)
                  at org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:359)
                  at org.apache.spark.sql.catalyst.trees.TreeNode.transformUp(TreeNode.scala:339)
                  at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformUp$1(TreeNode.scala:339)
                  at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$mapChildren$1(TreeNode.scala:408)
                  at org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:244)
  </code></pre>