title: Try to use sklearn library in Spark Structured Streaming
body: <p>I want to apply the label encoder function of sklearn.preprocessing on streaming data using Kafka and Spark structured streaming. So far the idea is:</p>
  
  <p>When I receive a batch of data from the Kafka source each time, I want to implement a function on that batch as follows: </p>
  
  <pre><code>def use_label_encoder(label_encoder, y):
     return label_encoder.transform(y) + 1
  
  to_transform_class_val = udf(use_label_encoder, IntegerType())
  </code></pre>
  
  <p>Here is the schema:</p>
  
  <pre><code>schema = StructType([
  StructField("sepal_length_in_cm", StringType()), \
  StructField("sepal_width_in_cm", StringType()), \
  StructField("petal_length_in_cm", StringType()), \
  StructField("petal_width_in_cm", StringType()), \
  StructField("class", StringType())
  ])
  
  df = df.selectExpr("CAST(value AS STRING)")
  df1 = df.select(from_json(df.value, schema).alias("json"))
  </code></pre>
  
  <p>when I try to defined label_encoder:</p>
  
  <pre><code>label_encoder = enc.fit(df1.select(to_upper("json.class")))
  </code></pre>
  
  <p>It gives an error "bad input shape"</p>
  
  <p>The equivalent code I used for non-streaming data is:</p>
  
  <pre><code>y = df['class'].values
  enc = LabelEncoder()
  label_encoder = enc.fit(y)
  y = label_encoder.transform(y) + 1
  </code></pre>
  
  <p>Could anyone please help me on how to apply sklearn method on streaming data?</p>