title: Split and filter the column of data frame in Spark
body: <p>I am working with apache spark. I have the following txt file. </p>
  
  <pre><code>05:49:56.604899 00:00:00:00:00:02 &gt; 00:00:00:00:00:03, ethertype IPv4 (0x0800), length 10202: 10.0.0.2.54880 &gt; 10.0.0.3.5001: Flags [.], seq 3641977583:3641987719, ack 129899328, win 58, options [nop,nop,TS val 432623 ecr 432619], length 10136
  05:49:56.604908 00:00:00:00:00:03 &gt; 00:00:00:00:00:02, ethertype IPv4 (0x0800), length 66: 10.0.0.3.5001 &gt; 10.0.0.2.54880: Flags [.], ack 10136, win 153, options [nop,nop,TS val 432623 ecr 432623], length 0
  05:49:56.604900 00:00:00:00:00:02 &gt; 00:00:00:00:00:03, ethertype IPv4 (0x0800), length 4410: 10.0.0.2.54880 &gt; 10.0.0.3.5001: Flags [P.], seq 10136:14480, ack 1, win 58, options [nop,nop,TS val 432623 ecr 432619], length 4344
  </code></pre>
  
  <p>Now I would like to extract the IPs and time stamp from the file. For example the output should become like below:</p>
  
  <pre><code>05:49: 56.604899 10.0.0.2 54880 10.0.0.3 5001
  05:49: 56.604908 10.0.0.3 5001 10.0.0.2 54880
  05:49: 56.604900 10.0.0.2 54880 10.0.0.3 5001
  </code></pre>
  
  <p>Here is the code that I used:</p>
  
  <pre><code>object ML_Test {
    def main(args: Array[String]) {
  
      val conf = new SparkConf().setAppName("saeed_test").setMaster("local[*]")
      val sc = new SparkContext(conf)
  
      val sqlContext = new SQLContext(sc)
  
      val customSchema = StructType(Array(
        StructField("column0", StringType, true),
        StructField("column1", StringType, true),
        StructField("column2", StringType, true)))
  
      val df = sqlContext.read
        .format("com.databricks.spark.csv")
        .option("header", "true") // Use first line of all files as header
        .schema(customSchema)
        .load("/Users/saeedtkh/Desktop/sharedsaeed/train.csv")
  
      val selectedData = df.select("column0", "column1", "column2")
      selectedData.write
        .format("com.databricks.spark.csv")
        .option("header", "true")
        .save("/Users/saeedtkh/Desktop/sharedsaeed/tempoutput.txt")
    }
  }
  </code></pre>
  
  <p>However I could just extract the following result. (I also could not apply split function here)</p>
  
  <pre><code>column0                                                  column1              column2
  05:29:59.546965 00:00:00:00:00:01 &gt; 00:00:00:00:00:03  ethertype IPv4 (0x0800) 05:29:59.546965 00:00:00:00:00:01 &gt; 00:00:00:00:00:05
  05:29:59.546986 00:00:00:00:00:01 &gt; 00:00:00:00:00:03  ethertype IPv4 (0x0800)  length 66: 10.0.0.1.5001 &gt; 10.0.0.3.43906: Flags [.]
  05:29:59.546986 00:00:00:00:00:01 &gt; 00:00:00:00:00:03  ethertype IPv4 (0x0800)  length 66: 10.0.0.1.5001 &gt; 10.0.0.3.43906: Flags [.]
  </code></pre>
  
  <p>Can you help me to modify this code to do the above result. Please help me.</p>
  
  <p><strong>Update1:</strong> 
  As I try to execute the answer number one, some fields can not recognize in my idea:
  [![enter image description here][1]][1]</p>
  
  <p>I added the following library and the problem get solved:</p>
  
  <pre><code>import org.apache.spark.sql.Row
  </code></pre>
  
  <p><strong>Update2:</strong>
  According to answer number one, as I ran the code in my idea, I got an empty folder as result. (Process finished with exit code 1)
  The errors are:</p>
  
  <pre><code>       17/05/24 09:45:52 ERROR Utils: Aborting task
  java.lang.ArrayIndexOutOfBoundsException: 2
      at ML_Test$$anonfun$2.apply(ML_Test.scala:28)
      at ML_Test$$anonfun$2.apply(ML_Test.scala:25)
      at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
      at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
      at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
      at org.apache.spark.sql.execution.datasources.DefaultWriterContainer$$anonfun$writeRows$1.apply$mcV$sp(WriterContainer.scala:254)
      at org.apache.spark.sql.execution.datasources.DefaultWriterContainer$$anonfun$writeRows$1.apply(WriterContainer.scala:252)
      at org.apache.spark.sql.execution.datasources.DefaultWriterContainer$$anonfun$writeRows$1.apply(WriterContainer.scala:252)
      at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1325)
      at org.apache.spark.sql.execution.datasources.DefaultWriterContainer.writeRows(WriterContainer.scala:258)
      at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelationCommand$$anonfun$run$1$$anonfun$apply$mcV$sp$1.apply(InsertIntoHadoopFsRelationCommand.scala:143)
      at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelationCommand$$anonfun$run$1$$anonfun$apply$mcV$sp$1.apply(InsertIntoHadoopFsRelationCommand.scala:143)
      at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70)
      at org.apache.spark.scheduler.Task.run(Task.scala:85)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
      at java.lang.Thread.run(Thread.java:745)
  17/05/24 09:45:52 ERROR DefaultWriterContainer: Task attempt attempt_201705240945_0000_m_000001_0 aborted.
  17/05/24 09:45:52 ERROR Executor: Exception in task 1.0 in stage 0.0 (TID 1)
  org.apache.spark.SparkException: Task failed while writing rows
      at org.apache.spark.sql.execution.datasources.DefaultWriterContainer.writeRows(WriterContainer.scala:261)
      at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelationCommand$$anonfun$run$1$$anonfun$apply$mcV$sp$1.apply(InsertIntoHadoopFsRelationCommand.scala:143)
      at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelationCommand$$anonfun$run$1$$anonfun$apply$mcV$sp$1.apply(InsertIntoHadoopFsRelationCommand.scala:143)
      at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70)
      at org.apache.spark.scheduler.Task.run(Task.scala:85)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
      at java.lang.Thread.run(Thread.java:745)
  Caused by: java.lang.ArrayIndexOutOfBoundsException: 2
      at ML_Test$$anonfun$2.apply(ML_Test.scala:28)
      at ML_Test$$anonfun$2.apply(ML_Test.scala:25)
      at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
      at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
      at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
      at org.apache.spark.sql.execution.datasources.DefaultWriterContainer$$anonfun$writeRows$1.apply$mcV$sp(WriterContainer.scala:254)
      at org.apache.spark.sql.execution.datasources.DefaultWriterContainer$$anonfun$writeRows$1.apply(WriterContainer.scala:252)
      at org.apache.spark.sql.execution.datasources.DefaultWriterContainer$$anonfun$writeRows$1.apply(WriterContainer.scala:252)
      at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1325)
      at org.apache.spark.sql.execution.datasources.DefaultWriterContainer.writeRows(WriterContainer.scala:258)
      ... 8 more
  </code></pre>