和 TensorFlow 集成¶
这篇指引会介绍如何在 Mars 里集成 TensorFlow 。
本指引基于 TensorFlow 2.0。
安装¶
如果你尝试在单机比如你的笔记本上使用 Mars，确保 TensorFlow 已经安装。
通过 pip 安装 TensorFlow：
pip install tensorflow
访问 TensorFlow 安装指引 获取更多信息
另一方面，如果你打算在集群中使用 Mars，确保 TensorFlow 在每个 worker 上安装。
准备数据¶
这里我们使用 ionosphere 数据集，点击链接下载数据。
编写 TensorFlow 脚本¶
现在我们创建一个命名为 tf_demo.py 的 Python 文件，它包含了 TensorFlow 的处理逻辑。
import os
import mars.dataframe as md
import tensorflow as tf
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import LabelEncoder
from tensorflow.keras import Sequential
from tensorflow.keras.layers import Dense
def prepare_data():
df = md.read_csv('ionosphere.data', header=None)
# split into input and output columns
X = df.iloc[:, :-1].to_tensor().astype('float32')
y = df.iloc[:, -1].to_tensor()
# convert Mars tensor to numpy ndarray
X, y = X.to_numpy(), y.to_numpy()
# encode string to integer
y = LabelEncoder().fit_transform(y)
# split into train and test datasets
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.33)
print(X_train.size, X_test.size, y_train.size, y_test.size)
return X_train, X_test, y_train, y_test
def get_model(n_features):
model = Sequential()
model.add(Dense(10, activation='relu', kernel_initializer='he_normal',
input_shape=(n_features,)))
model.add(Dense(8, activation='relu', kernel_initializer='he_normal'))
model.add(Dense(1, activation='sigmoid'))
# compile the model
model.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])
return model
def train():
X_train, X_test, y_train, y_test = prepare_data()
model = get_model(X_train.shape[1])
# fit model
model.fit(X_train, y_train, epochs=150, batch_size=32, verbose=0)
# evaluate
loss, acc = model.evaluate(X_test, y_test, verbose=0)
print('Test accuracy: %.3f' % acc)
if __name__ == '__main__':
if 'TF_CONFIG' in os.environ:
# distributed TensorFlow
multiworker_strategy = tf.distribute.experimental.MultiWorkerMirroredStrategy()
with multiworker_strategy.scope():
train()
else:
train()
Mars DataFrame 等模块可以直接在脚本里使用，以处理大规模数据和加速预处理。
通过 Mars 运行 TensorFlow 脚本¶
现在可以通过 run_tensorflow_script() 提交 TensorFlow 脚本。
In [1]: from mars.learn.contrib.tensorflow import run_tensorflow_script
In [2]: run_tensorflow_script('tf_demo.py', n_workers=1)
2020-04-28 15:40:38.284763: I tensorflow/core/platform/cpu_feature_guard.cc:142] Your CPU supports instructions that this TensorFlow binary was not compiled to use: AVX2 FMA
sh: sysctl: command not found
2020-04-28 15:40:38.301635: I tensorflow/compiler/xla/service/service.cc:168] XLA service 0x7fd29699c020 executing computations on platform Host. Devices:
2020-04-28 15:40:38.301656: I tensorflow/compiler/xla/service/service.cc:175] StreamExecutor device (0): Host, Default Version
2020-04-28 15:40:38.303779: I tensorflow/core/distributed_runtime/rpc/grpc_channel.cc:258] Initialize GrpcChannelCache for job worker -> {0 -> localhost:2221}
2020-04-28 15:40:38.304476: I tensorflow/core/distributed_runtime/rpc/grpc_server_lib.cc:365] Started server with target: grpc://localhost:2221
7990 3944 235 116
WARNING:tensorflow:`eval_fn` is not passed in. The `worker_fn` will be used if an "evaluator" task exists in the cluster.
WARNING:tensorflow:`eval_strategy` is not passed in. No distribution strategy will be used for evaluation.
WARNING:tensorflow:ModelCheckpoint callback is not provided. Workers will need to restart training if any fails.
WARNING:tensorflow:`eval_fn` is not passed in. The `worker_fn` will be used if an "evaluator" task exists in the cluster.
WARNING:tensorflow:`eval_strategy` is not passed in. No distribution strategy will be used for evaluation.
WARNING:tensorflow:ModelCheckpoint callback is not provided. Workers will need to restart training if any fails.
Test accuracy: 0.931
2020-04-28 15:40:45.906407: W tensorflow/core/common_runtime/eager/context.cc:290] Unable to destroy server_ object, so releasing instead. Servers don't support clean shutdown.
Out[2]: {'status': 'ok'}
分布式训练和推理¶
部署参考 在集群中部署 部分，在 Kubernetes 上运行参考 在 Kubernetes 中部署 部分。
从 tf_demo.py 中可以看出，Mars 会自动设置 TF_CONFIG 环境变量，其中包含集群和任务信息。因而，你不需要考虑如何规划分布式集群，只需要选择一个正确的 分布式策略。
一旦一个集群存在，你可以要么设置默认 session，训练和预测就会自动提交到集群，要么你可以通过 session=*** 显示指定运行的 session。
# A cluster has been configured, and web UI is started on <web_ip>:<web_port>
import mars
# set the session as the default one
sess = mars.new_session('http://<web_ip>:<web_port>')
# submitted to cluster by default
run_tensorflow_script('tf_demo.py', n_workers=1)
# Or, session could be specified as well
run_tensorflow_script('tf_demo.py', n_workers=1, session=sess)
使用 gen_tensorflow_dataset¶
你可以利用 gen_tensorflow_dataset() 将Mars类型数据（mars.tensor.Tensor，mars.dataframe.DataFrame，mars.dataframe.Series）转换为 tf.data.Dataset。该接口也支持 numpy.ndarray，pandas.DataFrame，pandas.Series。
In [1]: data = mt.tensor([[1, 2], [3, 4]])
In [2]: dataset = gen_tensorflow_dataset(data)
In [3]: list(dataset.as_numpy_iterator())
Out[3]: [array([1, 2]), array([3, 4])]
In [1]: data1 = mt.tensor([1, 2]); data2 = mt.tensor([3, 4]); data3 = mt.tensor([5, 6])
In [2]: dataset = gen_tensorflow_dataset((data1, data2, data3))
In [3]: list(dataset.as_numpy_iterator())
Out[3]: [(1, 3, 5), (2, 4, 6)]
现在，你可以用Mars预处理数据，然后将数据传到脚本中。
import mars.dataframe as md
from sklearn.preprocessing import LabelEncoder
from mars.learn.contrib.tensorflow import run_tensorflow_script
df = md.read_csv('ionosphere.data', header=None)
X = df.iloc[:, :-1].astype('float32')
y = df.iloc[:, -1]
y = LabelEncoder().fit_transform(y.execute().fetch())
X_train, X_test, y_train, y_test = train_test_split(X.execute(), y, test_size=0.33)
run_tensorflow_script(
"tf_demo.py", n_workers=2, data={'X_train': X_train, 'y_train': y_train,
'X_test':X_test, 'y_test': y_test}, session=sess)
tf_demo.py
import os
from mars.learn.contrib.tensorflow import gen_tensorflow_dataset
import tensorflow as tf
from tensorflow.keras import Sequential
from tensorflow.keras.layers import Dense
def get_model(n_features):
model = Sequential()
model.add(Dense(10, activation='relu', kernel_initializer='he_normal',
input_shape=(n_features,)))
model.add(Dense(8, activation='relu', kernel_initializer='he_normal'))
model.add(Dense(1, activation='sigmoid'))
# compile the model
model.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])
return model
def train(X_train, X_test, y_train, y_test):
model = get_model(X_train.shape[1])
db_train = gen_tensorflow_dataset((X_train, y_train))
db_train = db_train.batch(32)
db_test = gen_tensorflow_dataset((X_test, y_test))
db_test = db_test.batch(32)
# fit model
model.fit(db_train, epochs=150)
# evaluate
loss, acc = model.evaluate(db_test)
print('Test accuracy: %.3f' % acc)
if __name__ == '__main__':
X_train = globals()['X_train']
y_train = globals()['y_train']
X_test = globals()['X_test']
y_test = globals()['y_test']
if 'TF_CONFIG' in os.environ:
# distributed TensorFlow
multiworker_strategy = tf.distribute.experimental.MultiWorkerMirroredStrategy()
with multiworker_strategy.scope():
train(X_train, X_test, y_train, y_test)
else:
train(X_train, X_test, y_train, y_test)
结果：
Epoch 1/150
Epoch 1/150
1/Unknown - 1s 996ms/step - loss: 0.7825 - accuracy: 0.2500 1/Unknown - 1s 996ms/step - loss: 0.7825 - accura
6/Unknown - 3s 362ms/step - loss: 0.7388 - accuracy: 0.3438 6/Unknown - 3s 363ms/step - loss: 0.7388 - accura
7/Unknown - 3s 358ms/step - loss: 0.7404 - accuracy: 0.3259 7/Unknown - 3s 358ms/step - loss: 0.7404 - accura
8/Unknown - 3s 324ms/step - loss: 0.7368 - accuracy: 0.3277 8/Unknown - 3s 324ms/step - loss: 0.7368 - accura
8/8 [==============================] - 3s 324ms/step - loss: 0.7368 - accuracy: 0.3277
8/8 [==============================] - 3s 324ms/step - loss: 0.7368 - accuracy: 0.3277
Epoch 2/150
Epoch 2/150
8/8 [==============================] - ETA: 0s - loss: 0.6775 - accuracy: 0.49798/8 [==============================] - E
8/8 [==============================] - 3s 314ms/step - loss: 0.6775 - accuracy: 0.4979
8/8 [==============================] - 3s 314ms/step - loss: 0.6775 - accuracy: 0.4979
Epoch 3/150
Epoch 3/150
...
Epoch 150/150
Epoch 150/150
2/8 [======>.......................] - ETA: 2s - loss: 0.0210 - accuracy: 1.00002/8 [======>.......................] - E
3/8 [==========>...................] - ETA: 1s - loss: 0.0220 - accuracy: 1.00003/8 [==========>...................] - E
8/8 [==============================] - ETA: 0s - loss: 0.0319 - accuracy: 0.99578/8 [==============================] - E
8/8 [==============================] - 3s 351ms/step - loss: 0.0319 - accuracy: 0.9957
8/8 [==============================] - 3s 351ms/step - loss: 0.0319 - accuracy: 0.9957
. Consider either turning off auto-sharding or switching the auto_shard_policy to DATA to shard this dataset. You can do
this by creating a new `tf.data.Options()` object then setting `options.experimental_distribute.auto_shard_policy = Aut
oShardPolicy.DATA` before applying the options object to the dataset via `dataset.with_options(options)`.
4/Unknown - 3s 380ms/step - loss: 0.2354 - accuracy: 0.9138 4/Unknown - 3s 380ms/step - loss: 0.2354 - accura
4/4 [==============================] - 3s 381ms/step - loss: 0.2354 - accuracy: 0.9138
4/4 [==============================] - 3s 381ms/step - loss: 0.2354 - accuracy: 0.9138
Test accuracy: 0.914
Test accuracy: 0.914