为SparkDeepLearning添加NLP处理实现

前段时间研究了SDL项目，看到了Spark的宏大愿景，写了篇Spark新愿景：让深度学习变得更加易于使用。后面看了TFoS,感觉很是巧妙，写了一篇TensorFlowOnSpark 源码解析。这些项目都得益于Spark对python的支持，所以了解了下spark和python如何进行交互的，可参看此文PySpark如何设置worker的python命令。

虽然非常看好SDL,但是它存在几个明显的问题：

1. 进度慢的让人难以忍受。截止到目前为止，已经有26天没有新commit了。
2. 只做了图像相关的工作，没有任何NLP相关的工具使用。 参看其他人提的这个Issue: What would it take to generalize to non-image data?
3. 现有的分布式调参功能，基本不可用。参看我提的这个Issue: To Avoid collecting trainning data to driver and broadcasting them
4. 不支持分布式tranning. 参看我提的这个Issue: Is there any plan to port TensorframeOnSpark(From yahoo) ?

当然SDL的想法非常好：

1. 相比K8s + TF只是完成了分布式训练, SDL 把data process ,data training,data inference 三者给完全衔接了。
2. 提供了一个很好的编程模型，以sk-learn/Mllib的方式完成模型的训练，对于工作效率提升明显。
3. 分布式模型训练，分布式模型超参数tunning, 分别解决了训练数据量大的问题，参数探索的问题。

因为我司以NLP为主，所以我提供了一个deep learning auto-encoder的一个demo,展现SDL的能力。顺带通过引入Kafka解决了
&＃8220;分布式模型超参数tunning&＃8221;在实际场景不可用的问题。有时间会完成和TFoS的集成。

我这里写了一个单元测试(python/tests/transformers/tf_text_test.py)：

class TFTextTransformerTest(SparkDLTestCase):
def test_loadText(self):
input_col = "text"
output_col = "sentence_matrix"
documentDF = self.session.createDataFrame([
("Hi I heard about Spark", 1),
("I wish Java could use case classes", 0),
("Logistic regression models are neat", 2)
], ["text", "preds"])
# transform text column to sentence_matrix column which contains 2-D array.
transformer = TFTextTransformer(
inputCol=input_col, outputCol=output_col)
df = transformer.transform(documentDF)
# create a estimator to training where map_fun contains tensorflow's code
estimator = TFTextFileEstimator(inputCol="sentence_matrix", outputCol="sentence_matrix", labelCol="preds",
kafkaParam={"host": "127.0.0.1", "topic": "test", "group_id": "sdl_1"},
fitParam=[{"epochs": 5, "batch_size": 64}, {"epochs": 5, "batch_size": 1}],
mapFnParam=map_fun)
estimator.fit(df).collect()


TFTextTransformer 主要是把任意文本转化为一个二维矩阵，一行代表一个词汇，每个词汇都是word embedding的形态。该Transformer本质是做featurize的工作，2-D array 是能够直接被包括CNN,LSTM等算法操作的格式。 我这里简要介绍下TFTextTransformer的处理流程：

1. 获取输入列，然后使用word2vec对数据进行训练，得到每个词的word embedding，最后作为一个map(word, vector) 广播出去
2. 将input_col列的句子转化为一个2-D array作为outputCol
3. 添加一些常数列到新的DataFrame里，比如vocab_size(词汇数目)，embedding_size(词向量大小)。
4. 返回新DataFrame

TFTextFileEstimator 完成训练过程，具体流程为：

1. TFTextFileEstimator 将TFTextTransformer的每一条数据序列化后写入Kafka
2. 根据fitParams （也就是你设置的超参数组合）长度，启动对应个数的tensorflow实例
3. 为tensorflow实例从kafka拉去数据，并且提供一个_read_data函数句柄给tensorflow程序。
4. 调用你编写的tf程序，完成训练。

额外引入kafka的原因是因为，每个tensorflow实例都需要消费全量的数据，一个简单的做法是把数据collect到driver端然后broadcast出去，但是实际上行不通，所以将数据集中放在kafka。

map_fun 是一个函数，这里你完全可以使用keras/tensorflow 构建模型，并且调用_read_data获取数据，以及通过args获得必要的参数，具体代码(python/sparkdl/tf_fun.py)：

def map_fun(_read_data, **args):
import tensorflow as tf
EMBEDDING_SIZE = args["embedding_size"]
feature = args['feature']
label = args['label']
params = args['params']['fitParam']
SEQUENCE_LENGTH = 64
def feed_dict(batch):
# Convert from dict of named arrays to two numpy arrays of the proper type
features = []
for i in batch:
features.append(i['sentence_matrix'])
# print("{} {}".format(feature, features))
return features
encoder_variables_dict = {
"encoder_w1": tf.Variable(
tf.random_normal([SEQUENCE_LENGTH * EMBEDDING_SIZE, 256]), name="encoder_w1"),
"encoder_b1": tf.Variable(tf.random_normal([256]), name="encoder_b1"),
"encoder_w2": tf.Variable(tf.random_normal([256, 128]), name="encoder_w2"),
"encoder_b2": tf.Variable(tf.random_normal([128]), name="encoder_b2")
}


_read_data 可以获取spark dataframe的数据，典型用法如下：

for i in range(params.epochs):
print("epoll {}".format(i))
for data in _read_data(max_records=params.batch_size):
batch_data = feed_dict(data)
sess.run(train_step, feed_dict={input_x: batch_data})
sess.close()


这里，你核心关注如何构建网络，数据处理的工作前面的transformer已经帮你完成。

详细代码参看： https://github.com/allwefantasy/spark-deep-learning/tree/nlp-support