pyspark：スパース行列をscipyスパース行列にする

Question

私は短い文の列とカテゴリ変数を持つ列を持つスパークデータフレームを持っています。文章でtf-idfを実行したい場合は、カテゴリ変数のone-hot-encodingを入力してから、サイズがはるかに小さくなるとドライバのスパース行列に出力します（scikit-learnモデルの場合）。

sparse形式でsparkからデータを取り出す最も良い方法は何ですか？疎ベクトル上にtoArray()メソッドしかなく、numpy配列を出力するようです。しかし、ドキュメントではscipyの疎の配列があります。can be used in the place of spark sparse arrays.

tf_idfの値は実際には疎な配列の列であることにも注意してください。理想的には、これらすべての機能を1つの大きなスパース行列にまとめるとよいでしょう。

Answer 1

次のように一つの可能​​な解決策を表すことができる。

• 変換機能をRDDと抽出ベクター：

  from pyspark.ml.linalg import SparseVector 
  from operator import attrgetter 
  
  df = sc.parallelize([ 
      (SparseVector(3, [0, 2], [1.0, 3.0]),), 
      (SparseVector(3, [1], [4.0]),) 
  ]).toDF(["features"]) 
  
  features = df.rdd.map(attrgetter("features")) 
  

• 追加行インデックス：

  indexed_features = features.zipWithIndex() 
  

• に平坦タプルのRDD (i, j, value)：

  def explode(row): 
      vec, i = row 
      for j, v in zip(vec.indices, vec.values): 
       yield i, j, v 
  
  entries = indexed_features.flatMap(explode) 
  

• 収集し、再構築：

  row_indices, col_indices, data = zip(*entries.collect()) 
  

• コンピュート形状：

  shape = (
      df.count(), 
      df.rdd.map(attrgetter("features")).first().size 
  ) 
  

• スパース行列を作成します。

  from scipy.sparse import csr_matrix 
  
  mat = csr_matrix((data, (row_indices, col_indices)), shape=shape) 
  

• 迅速正気を

  matrix([[ 1., 0., 3.], 
       [ 0., 4., 0.]]) 
  

もう一つ：期待される結果と

mat.todense() 

：チェック

• マトリックスにfeaturesの各行を変換：

  import numpy as np 
  
  def as_matrix(vec): 
      data, indices = vec.values, vec.indices 
      shape = 1, vec.size 
      return csr_matrix((data, indices, np.array([0, vec.values.size])), shape) 
  
  mats = features.map(as_matrix) 
  

• とvstackを減らす：

  from scipy.sparse import vstack 
  
  mat = mats.reduce(lambda x, y: vstack([x, y])) 
  

  またはcollectとvstack

  mat = vstack(mats.collect()) 
  

を