title: How to split input file name and add specific value in the spark data frame column
body: <p>This is how i load my csv file in spark data frame </p>
  
  <pre><code>val sqlContext = new org.apache.spark.sql.SQLContext(sc)
  import sqlContext.implicits._
  
  import org.apache.spark.{ SparkConf, SparkContext }
  import java.sql.{Date, Timestamp}
  import org.apache.spark.sql.Row
  import org.apache.spark.sql.types._
  import org.apache.spark.sql.functions.udf
  
  
  
  val get_cus_val = spark.udf.register("get_cus_val", (filePath: String) =&gt; filePath.split("\\.")(4))
  
  val df1With_ = df.toDF(df.columns.map(_.replace(".", "_")): _*)
  val column_to_keep = df1With_.columns.filter(v =&gt; (!v.contains("^") &amp;&amp; !v.contains("!") &amp;&amp; !v.contains("_c"))).toSeq
  val df1result = df1With_.select(column_to_keep.head, column_to_keep.tail: _*)
  val df1Final=df1result.withColumn("DataPartition", lit(null: String))
  </code></pre>
  
  <p>This is example of one of my input file name .</p>
  
  <pre><code>Fundamental.FinancialLineItem.FinancialLineItem.SelfSourcedPrivate.CUS.1.2017-09-07-1056.Full
  
  Fundamental.FinancialLineItem.FinancialLineItem.Japan.CUS.1.2017-09-07-1056.Full.txt
  </code></pre>
  
  <p>Now i want to read this file and split it with "." operator and then add CUS as new column in place of DataPartition .</p>
  
  <p>Can i do it without any UDF .</p>
  
  <p>Here is the schema of existing data frame </p>
  
  <pre><code>root
   |-- LineItem_organizationId: long (nullable = true)
   |-- LineItem_lineItemId: integer (nullable = true)
   |-- StatementTypeCode: string (nullable = true)
   |-- LineItemName: string (nullable = true)
   |-- LocalLanguageLabel: string (nullable = true)
   |-- FinancialConceptLocal: string (nullable = true)
   |-- FinancialConceptGlobal: string (nullable = true)
   |-- IsDimensional: boolean (nullable = true)
   |-- InstrumentId: string (nullable = true)
   |-- LineItemSequence: string (nullable = true)
   |-- PhysicalMeasureId: string (nullable = true)
   |-- FinancialConceptCodeGlobalSecondary: string (nullable = true)
   |-- IsRangeAllowed: boolean (nullable = true)
   |-- IsSegmentedByOrigin: boolean (nullable = true)
   |-- SegmentGroupDescription: string (nullable = true)
   |-- SegmentChildDescription: string (nullable = true)
   |-- SegmentChildLocalLanguageLabel: string (nullable = true)
   |-- LocalLanguageLabel_languageId: integer (nullable = true)
   |-- LineItemName_languageId: integer (nullable = true)
   |-- SegmentChildDescription_languageId: integer (nullable = true)
   |-- SegmentChildLocalLanguageLabel_languageId: integer (nullable = true)
   |-- SegmentGroupDescription_languageId: integer (nullable = true)
   |-- SegmentMultipleFundbDescription: string (nullable = true)
   |-- SegmentMultipleFundbDescription_languageId: integer (nullable = true)
   |-- IsCredit: boolean (nullable = true)
   |-- FinancialConceptLocalId: integer (nullable = true)
   |-- FinancialConceptGlobalId: integer (nullable = true)
   |-- FinancialConceptCodeGlobalSecondaryId: string (nullable = true)
   |-- FFAction: string (nullable = true)
  </code></pre>
  
  <p>Updating the code after suggested answer </p>
  
  <pre><code>    val sqlContext = new org.apache.spark.sql.SQLContext(sc)
      import sqlContext.implicits._
  
      import org.apache.spark.{ SparkConf, SparkContext }
      import java.sql.{Date, Timestamp}
      import org.apache.spark.sql.Row
      import org.apache.spark.sql.types._
      import org.apache.spark.sql.functions.udf
      import org.apache.spark.sql.functions.{input_file_name, regexp_extract}
  
  spark.udf.register("get_cus_val", (filePath: String) =&gt; filePath.split("\\.")(4))
  
  import org.apache.spark.sql.functions.input_file_name
  
  val df = sqlContext.read.format("csv").option("header", "true").option("delimiter", "|").option("inferSchema","true").load("s3://trfsdisu/SPARK/FinancialLineItem/MAIN")
  
  val df1With_ = df.toDF(df.columns.map(_.replace(".", "_")): _*)
  val column_to_keep = df1With_.columns.filter(v =&gt; (!v.contains("^") &amp;&amp; !v.contains("!") &amp;&amp; !v.contains("_c"))).toSeq
  val df1result = df1With_.select(column_to_keep.head, column_to_keep.tail: _*)
  
  df1result.withColumn("cus_val", get_cus_val(input_file_name))
  
  df1result.printSchema()
  </code></pre>