前段时间研究了SDL项目,看到了Spark的宏大愿景,写了篇Spark新愿景:让深度学习变得更加易于使用。后面看了TFoS,感觉很是巧妙,写了一篇TensorFlowOnSpark 源码解析。这些项目都得益于Spark对python的支持,所以了解了下spark和python如何进行交互的,可参看此文PySpark如何设置worker的python命令。
虽然非常看好SDL,但是它存在几个明显的问题:
当然SDL的想法非常好:
因为我司以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的处理流程:
TFTextFileEstimator 完成训练过程,具体流程为:
额外引入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