为Spark Deep Learning 添加NLP处理实现

2023年3月26日 180次阅读 来源: 祝威廉

前言

前段时间研究了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解决了
“分布式模型超参数tunning”在实际场景不可用的问题。有时间会完成和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

    原文作者：祝威廉
    原文地址: https://www.jianshu.com/p/6e4005fb9304
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。
点赞