title: (KNN ) row compute use outer DataFrame on pyspark
body: <p><strong>question</strong></p>
  
  <p><strong>my data structure is like this:</strong></p>
  
  <pre><code>
  train_info:(over 30000 rows)
  ----------
  odt:string (unique)
  holiday_type:string
  od_label:string
  array:array&lt;double&gt; (with variable length depend on different odt and holiday_type )
  useful_index:array&lt;int&gt;  (length same as vectors)
  ...(other not important cols)
  
  label_data:(over 40000 rows)
  ----------
  holiday_type:string
  od_label: string
  l_origin_array:array&lt;double&gt; (with variable length)
  ...(other not important cols)
  
  my expected result is like this(length same with train_info):
  --------------
  odt:string
  holiday_label:string
  od_label:string
  prediction:int
  </code></pre>
  
  <p><strong>my solution is like this:</strong></p>
  
  <pre><code>if __name__=='__main __'
          loop_item = train_info.collect()
          result = knn_for_loop(spark, loop_item,train_info.schema,label_data)
          ----- do something -------
  
  </code></pre>
  
  <pre><code>def knn_for_loop(spark, predict_list, schema, label_data):
      result = list()
      for i in predict_list:
          # turn this Row col to Data Frame  and  join  on  label data 
          # across to this row data pick label data array data
          predict_df = spark.sparkContext.parallelize([i]).toDF(schema) \
              .join(label_data, on=['holiday_type', "od_label"], how='left') \
              .withColumn("l_array",
                       UDFuncs.value_from_array_by_index(f.col('l_origin_array'), f.col("useful_index"))) \
              .toPandas()
          # pandas execute
          train_x = predict_df.l_array.values
          train_y = predict_df.label.values
          test_x = predict_df.array.values[0]
          test_y = KNN(train_x, train_y, test_x)
  
          result.append((i['odt'], i['holiday_type'], i['od_label'], test_y))
      return result
  </code></pre>
  
  <p>it's worked but is really slow, I estimate each row need 18s.</p>
  
  <p>in R language  I can do this easily using <em>do</em> function:</p>
  
  <p>train_info%>%group_by(odt)%>%do(.,knn_loop,label_data)</p>
  
  <p><strong>something my tries</strong></p>
  
  <p>I tried to join them before use,and query them when I compute, but the data is too large to run (these two df have 400 million rows after join and It takes up 180 GB disk space on hive and query really slowly).
  I tried to use pandas_udf, but it only allows one pd.data.frame parameter and slow).</p>
  
  <p>I tried to use UDF, but UDF can't receive data frame obj.  </p>
  
  <p>I tried to use spark-knn package  ,but I run with error,may be my  offline 
   installation is  wrong .</p>
  
  <p>thanks for your help.</p>