title: How to convert Spark Streaming data into Spark DataFrame
body: <p>So far, Spark hasn't created the DataFrame for streaming data, but when I am doing anomalies detection, it is more convenient and faster to use DataFrame for data analysis. I have done this part, but when I try to do real time anomalies detection using streaming data, the problems appeared. I tried several ways and still could not convert DStream to DataFrame, and cannot convert the RDD inside of DStream into DataFrame either.</p>
  
  <p>Here's part of my latest version of the code:</p>
  
  <pre><code>import sys
  import re
  
  from pyspark import SparkContext
  from pyspark.sql.context import SQLContext
  from pyspark.sql import Row
  from pyspark.streaming import StreamingContext
  from pyspark.mllib.clustering import KMeans, KMeansModel, StreamingKMeans
  from pyspark.sql.functions import *
  from pyspark.sql.types import *
  from pyspark.sql.functions import udf
  import operator
  
  
  sc = SparkContext(appName="test")
  ssc = StreamingContext(sc, 5)
  sqlContext = SQLContext(sc)
  
  model_inputs = sys.argv[1]
  
  def streamrdd_to_df(srdd):
      sdf = sqlContext.createDataFrame(srdd)
      sdf.show(n=2, truncate=False)
      return sdf
  
  def main():
      indata = ssc.socketTextStream(sys.argv[2], int(sys.argv[3]))
      inrdd = indata.map(lambda r: get_tuple(r))
      Features = Row('rawFeatures')
      features_rdd = inrdd.map(lambda r: Features(r))
      features_rdd.pprint(num=3)
      streaming_df = features_rdd.flatMap(streamrdd_to_df)
  
      ssc.start()
      ssc.awaitTermination()
  
  if __name__ == "__main__":
      main()
  </code></pre>
  
  <p>As you can see in the main() function, when I am reading the input streaming data using ssc.socketTextStream() method, it generates DStream, then I tried to convert each individual in DStream into Row, hoping I could convert the data into DataFrame later.</p>
  
  <p>If I use ppprint() to print out features_rdd here, it works, which makes me think, each individual in features_rdd is a batch of RDD while the whole features_rdd is a DStream.</p>
  
  <p>Then I created streamrdd_to_df() method and hoped to convert each batch of RDD into dataframe, it gives me the error, showing:</p>
  
  <blockquote>
    <p>ERROR StreamingContext: Error starting the context, marking it as stopped
    java.lang.IllegalArgumentException: requirement failed: No output operations registered, so nothing to execute</p>
  </blockquote>
  
  <p>Is there any thought about how can I do DataFrame operations on Spark streaming data?</p>