碼上歡樂
相關內容    簡體    繁體

大數據下基於Tensorflow框架的深度學習示例教程

本文轉載自   查看原文   2019-04-23 17:53   685    機器人編程/ 人工智能/ 大數據下基於Tensorflow框架的深度學習示例教程

近幾年，信息時代的快速發展產生了海量數據，誕生了無數前沿的大數據技術與應用。在當今大數據時代的產業界，商業決策日益基於數據的分析作出。當數據膨脹到一定規模時，基於機器學習對海量復雜數據的分析更能產生較好的價值，而深度學習在大數據場景下更能揭示數據內部的邏輯關系。本文就以大數據作為場景，通過自底向上的教程詳述在大數據架構體系中如何應用深度學習這一技術。大數據架構中采用的是hadoop系統以及Kerberos安全認證，深度學習采用的是分布式的Tensorflow架構，hadoop解決了大數據的存儲問題，而分布式Tensorflow解決了大數據訓練的問題。本教程是我們團隊在開發基於深度學習的實時欺詐預警服務時，部署深度學習這一模塊時總結出的經驗，感興趣的歡迎深入交流。

安裝Tensorflow

我們安裝Tensorflow選擇的是Centos7，因為Tensorflow需要使用GNU發布的1.5版本的libc庫，Centos6系統並不適用該版本庫而被拋棄。對於如何聯網在線安裝Tensorflow，官網有比較詳盡的教程。本教程着重講一下網上資料較少的離線安裝方式，系統的安裝更需要在意的是各軟件版本的一致性，下面教程也是解決了很多版本不一致的問題后給出的一個方案。首先我們先將整個系統搭建起來吧。

1.安裝編程語言Python3.5：在官網下載軟件並解壓后執行如下安裝命令：

  1. ./configure
  2. make
  3. make test
  4. sudo make install

2.安裝基於Python的科學計算包python-numpy：在官網下載軟件並解壓后執行如下安裝命令：

python setup.py install

3.安裝Python模塊管理的工具wheel：在官網下載軟件后執行如下安裝命令：

pip install wheel-0.30.0a0-py2.py3-none-any.whl

4.安裝自動下載、構建、安裝和管理 python 模塊的工具setuptools：在官網下載軟件並解壓后執行如下安裝命令：

python setup.py install

5.安裝Python開發包python-devel：在官網下載軟件后執行如下安裝命令：

sudo rpm -i --nodeps python3-devel-3.5.2-4.fc25.x86_64.rpm

6.安裝Python包安裝管理工具six：在官網下載軟件后執行如下安裝命令：

sudo pip install six-1.10.0-py2.py3-none-any.whl

7.安裝Java 開發環境JDK8：在官網下載軟件並解壓后執行如下移動命令：

mv java1.8 /usr/local/software/jdk

設置JDK的環境變量，編輯文件 .bashrc，加入下面內容

  1. export JAVA_HOME=/usr/local/software/jdk
  2. export JRE_HOME=${JAVA_HOME}/jre
  3. export CLASSPATH=$CLASSPATH:${JAVA_HOME}/lib:${JRE_HOME}/lib
  4. export PATH=$PATH:${JAVA_HOME}/bin

進行Java版本的切換，選擇對應的版本

  1. sudo update-alternatives --config java
  2. sudo update-alternatives --config javac

8.安裝Bazel：Bazel是一個類似於Make的工具，是Google為其內部軟件開發的特點量身定制的工具，構建Tensorflow項目。在官網下載后執行如下安裝命令：

  1. chmod +x bazel-0.4.3-installer-linux-x86_64.sh
  2. ./bazel-0.4.3-installer-linux-x86_64.sh –user

9.安裝Tensorflow：在官網下載軟件后執行如下安裝命令：

pip install --upgrade tensorflow-0.12.1-cp35-cp35m-linux_x86_64.whl

Tensorflow訪問HDFS的部署

1.首先安裝Hadoop客戶端，在官網下載后執行下面解壓移動命令：

  1. tar zxvf hadoop- 2.6.0.tar.gz
  2. mv hadoop- 2.6.0.tar.gz /usr/local/software/Hadoop

進行環境變量的配置/etc/profile，加入如下內容

  1. export PATH=$PATH:/usr/local/software/hadoop/bin
  2. export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:$JAVA_HOME/jre/lib/amd64/server
  3. export HADOOP_HOME=/usr/local/software/hadoop
  4. export HADOOP_HDFS_HOME=/usr/local/software/hadoop

配置完后進行配置更新source /etc/profile

2.其次，安裝完客戶端后，配置自己的hadoop集群環境文件。

Tensorflow與Kerberos驗證的部署

在Tesorflow0.12版本中已經支持了Kerberos驗證，本機只要配置好Kerberos文件即可使用。該文中不詳述Kerberos的配置內容，羅列一下相關的配置流程。

  • 首先在/etc/krb5.conf文件中進行服務器跟驗證策略的配置；
  • 然后在Kerberos服務端生成一個用戶文件傳至本機；
  • 最后進行Kerberos客戶端的權限認證並設置定時任務。

大數據場景下基於分布式Tensorflow的深度學習示例

一、進行數據格式的轉換

本文的示例是做的MNIST數據的識別模型，為了更好的讀取數據更好的利用內存，我們將本地GZ文件轉換成Tensorflow的內定標准格式TFRecord，然后再將轉換后的文件上傳到HDFS存儲。在實際應用中，我們實際利用Spark做了大規模格式轉換的處理程序。我們對本地數據處理的相應的轉換代碼為：

  1. from __future__ import absolute_import
  2. from __future__ import division
  3. from __future__ import print_function
  4. import argparse
  5. import os
  6. import tensorflow as tf
  7. from tensorflow.contrib.learn.python.learn.datasets import mnist
  8. SOURCE_URL = 'http://yann.lecun.com/exdb/mnist/'
  9. TRAIN_IMAGES = 'train-images-idx3-ubyte.gz' # MNIST filenames
  10. TRAIN_LABELS = 'train-labels-idx1-ubyte.gz'
  11. TEST_IMAGES = 't10k-images-idx3-ubyte.gz'
  12. TEST_LABELS = 't10k-labels-idx1-ubyte.gz'
  13. FLAGS = None
  14. def _int64_feature(value):
  15. return tf.train.Feature(int64_list=tf.train.Int64List(value=[value]))
  16.  
  17. def _bytes_feature(value):
  18. return tf.train.Feature(bytes_list=tf.train.BytesList(value=[value]))
  19.  
  20. def convert_to(data_set, name):
  21. images = data_set.images
  22. labels = data_set.labels
  23. num_examples = data_set.num_examples
  24. if images.shape[0] != num_examples:
  25. raise ValueError('Images size %d does not match label size %d.' %
  26. (images.shape[ 0], num_examples))
  27. rows = images.shape[ 1]
  28. cols = images.shape[ 2]
  29. depth = images.shape[ 3]
  30. filename = os.path.join(FLAGS.directory, name + '.tfrecords')
  31. print( 'Writing', filename)
  32. writer = tf.python_io.TFRecordWriter(filename)
  33. for index in range(num_examples):
  34. image_raw = images[index].tostring()
  35. example = tf.train.Example(features=tf.train.Features(feature={
  36. 'height': _int64_feature(rows),
  37. 'width': _int64_feature(cols),
  38. 'depth': _int64_feature(depth),
  39. 'label': _int64_feature(int(labels[index])),
  40. 'image_raw': _bytes_feature(image_raw)}))
  41. writer.write(example.SerializeToString())
  42. writer.close()
  43.  
  44. def main(argv):
  45. # Get the data.
  46. data_sets = mnist.read_data_sets(FLAGS.directory,
  47. dtype=tf.uint8,
  48. reshape= False,
  49. validation_size=FLAGS.validation_size)
  50. # Convert to Examples and write the result to TFRecords.
  51. convert_to(data_sets.train, 'train')
  52. convert_to(data_sets.validation, 'validation')
  53. convert_to(data_sets.test, 'test')
  54. if __name__ == '__main__':
  55. parser = argparse.ArgumentParser()
  56. parser.add_argument(
  57. '--directory',
  58. type=str,
  59. default= '/tmp/data',
  60. help= 'Directory to download data files and write the converted result'
  61. )
  62. parser.add_argument(
  63. '--validation_size',
  64. type=int,
  65. default= 5000,
  66. help= """\
  67. Number of examples to separate from the training data for the validation
  68. set.\
  69. """
  70. )
  71. FLAGS = parser.parse_args()
  72. tf.app.run()

二、Tensorflow讀取HDFS數據的設置

文中前面內容介紹了HDFS的配置以及將數據轉換后存儲到HDFS，Tensorflow讀取HDFS時只需要簡單的兩步，首先執行項目時需要加入環境前綴：

CLASSPATH=$($HADOOP_HDFS_HOME/bin/hadoop classpath --glob) python example.py

其次讀取數據時，需要在數據的路徑前面加入HDFS前綴，比如：

hdfs://default/user/data/example.txt

三、分布式模型的示例代碼

該示例代碼是讀取HDFS上的MNIST數據，建立相應的server與work集群構建出一個三層的深度網絡，包含兩層卷積層以及一層SoftMax層。代碼如下：

  1. from __future__ import print_function
  2. import math
  3. import os
  4. import tensorflow as tf
  5. flags = tf.app.flags
  6. # Flags for configuring the task
  7. flags.DEFINE_string( "job_name", None, "job name: worker or ps")
  8. flags.DEFINE_integer( "task_index", 0,
  9. "Worker task index, should be >= 0. task_index=0 is "
  10. "the chief worker task the performs the variable "
  11. "initialization")
  12. flags.DEFINE_string( "ps_hosts", "",
  13. "Comma-separated list of hostname:port pairs")
  14. flags.DEFINE_string( "worker_hosts", "",
  15. "Comma-separated list of hostname:port pairs")
  16. # Training related flags
  17. flags.DEFINE_string( "data_dir", None,
  18. "Directory where the mnist data is stored")
  19. flags.DEFINE_string( "train_dir", None,
  20. "Directory for storing the checkpoints")
  21. flags.DEFINE_integer( "hidden1", 128,
  22. "Number of units in the 1st hidden layer of the NN")
  23. flags.DEFINE_integer( "hidden2", 128,
  24. "Number of units in the 2nd hidden layer of the NN")
  25. flags.DEFINE_integer( "batch_size", 100, "Training batch size")
  26. flags.DEFINE_float( "learning_rate", 0.01, "Learning rate")
  27. FLAGS = flags.FLAGS
  28. TRAIN_FILE = "train.tfrecords"
  29. NUM_CLASSES = 10
  30. IMAGE_SIZE = 28
  31. IMAGE_PIXELS = IMAGE_SIZE * IMAGE_SIZE
  32.  
  33. def inference(images, hidden1_units, hidden2_units):
  34. with tf.name_scope('hidden1'):
  35. weights = tf.Variable(
  36. tf.truncated_normal([IMAGE_PIXELS, hidden1_units],
  37. stddev= 1.0 / math.sqrt(float(IMAGE_PIXELS))),name='weights')
  38. biases = tf.Variable(tf.zeros([hidden1_units]),name= 'biases')
  39. hidden1 = tf.nn.relu(tf.matmul(images, weights) + biases)
  40. with tf.name_scope('hidden2'):
  41. weights = tf.Variable(
  42. tf.truncated_normal([hidden1_units, hidden2_units],
  43. stddev= 1.0 / math.sqrt(float(hidden1_units))),
  44. name= 'weights')
  45. biases = tf.Variable(tf.zeros([hidden2_units]),
  46. name= 'biases')
  47. hidden2 = tf.nn.relu(tf.matmul(hidden1, weights) + biases)
  48. with tf.name_scope('softmax_linear'):
  49. weights = tf.Variable(
  50. tf.truncated_normal([hidden2_units, NUM_CLASSES],
  51. stddev= 1.0 / math.sqrt(float(hidden2_units))),name='weights')
  52. biases = tf.Variable(tf.zeros([NUM_CLASSES]),name= 'biases')
  53. logits = tf.matmul(hidden2, weights) + biases
  54. return logits
  55.  
  56. def lossFunction(logits, labels):
  57. labels = tf.to_int64(labels)
  58. cross_entropy = tf.nn.sparse_softmax_cross_entropy_with_logits(
  59. logits, labels, name= 'xentropy')
  60. loss = tf.reduce_mean(cross_entropy, name= 'xentropy_mean')
  61. return loss
  62.  
  63. def training(loss, learning_rate):
  64. tf.summary.scalar(loss.op.name, loss)
  65. optimizer = tf.train.GradientDescentOptimizer(learning_rate)
  66. global_step = tf.Variable( 0, name='global_step', trainable=False)
  67. train_op = optimizer.minimize(loss, global_step=global_step)
  68. return train_op
  69.  
  70. def read_and_decode(filename_queue):
  71. reader = tf.TFRecordReader()
  72. _, serialized_example = reader.read(filename_queue)
  73. features = tf.parse_single_example(
  74. serialized_example,
  75. # Defaults are not specified since both keys are required.
  76. features={
  77. 'image_raw': tf.FixedLenFeature([], tf.string),
  78. 'label': tf.FixedLenFeature([], tf.int64),
  79. })
  80.  
  81. # Convert from a scalar string tensor (whose single string has
  82. # length mnist.IMAGE_PIXELS) to a uint8 tensor with shape
  83. # [mnist.IMAGE_PIXELS].
  84. image = tf.decode_raw(features[ 'image_raw'], tf.uint8)
  85. image.set_shape([IMAGE_PIXELS])
  86. image = tf.cast(image, tf.float32) * ( 1. / 255) - 0.5
  87. # Convert label from a scalar uint8 tensor to an int32 scalar.
  88. label = tf.cast(features[ 'label'], tf.int32)
  89. return image, label
  90.  
  91. def inputs(batch_size):
  92. """Reads input data.
  93.  
  94. Args:
  95. batch_size: Number of examples per returned batch.
  96. Returns:
  97. A tuple (images, labels), where:
  98. * images is a float tensor with shape [batch_size, mnist.IMAGE_PIXELS]
  99. in the range [-0.5, 0.5].
  100. * labels is an int32 tensor with shape [batch_size] with the true label,
  101. a number in the range [0, mnist.NUM_CLASSES).
  102. """
  103. filename = os.path.join(FLAGS.data_dir, TRAIN_FILE)
  104.  
  105. with tf.name_scope('input'):
  106. filename_queue = tf.train.string_input_producer([filename])
  107. # Even when reading in multiple threads, share the filename
  108. # queue.
  109. image, label = read_and_decode(filename_queue)
  110. # Shuffle the examples and collect them into batch_size batches.
  111. # (Internally uses a RandomShuffleQueue.)
  112. # We run this in two threads to avoid being a bottleneck.
  113. images, sparse_labels = tf.train.shuffle_batch(
  114. [image, label], batch_size=batch_size, num_threads= 2,
  115. capacity= 1000 + 3 * batch_size,
  116. # Ensures a minimum amount of shuffling of examples.
  117. min_after_dequeue= 1000)
  118. return images, sparse_labels
  119.  
  120. def device_and_target():
  121. # If FLAGS.job_name is not set, we're running single-machine TensorFlow.
  122. # Don't set a device.
  123. if FLAGS.job_name is None:
  124. raise ValueError("Must specify an explicit `job_name`")
  125. # Otherwise we're running distributed TensorFlow.
  126. print( "Running distributed training")
  127. if FLAGS.task_index is None or FLAGS.task_index == "":
  128. raise ValueError("Must specify an explicit `task_index`")
  129. if FLAGS.ps_hosts is None or FLAGS.ps_hosts == "":
  130. raise ValueError("Must specify an explicit `ps_hosts`")
  131. if FLAGS.worker_hosts is None or FLAGS.worker_hosts == "":
  132. raise ValueError("Must specify an explicit `worker_hosts`")
  133. cluster_spec = tf.train.ClusterSpec({
  134. "ps": FLAGS.ps_hosts.split(","),
  135. "worker": FLAGS.worker_hosts.split(","),
  136. })
  137. server = tf.train.Server(
  138. cluster_spec, job_name=FLAGS.job_name, task_index=FLAGS.task_index)
  139. return (
  140. cluster_spec,
  141. server,
  142. )
  143.  
  144. def main(unused_argv):
  145. if FLAGS.data_dir is None or FLAGS.data_dir == "":
  146. raise ValueError("Must specify an explicit `data_dir`")
  147. if FLAGS.train_dir is None or FLAGS.train_dir == "":
  148. raise ValueError("Must specify an explicit `train_dir`")
  149. cluster_spec, server = device_and_target()
  150. if FLAGS.job_name == "ps":
  151. server.join()
  152. elif FLAGS.job_name == "worker":
  153. with tf.device(tf.train.replica_device_setter(worker_device = "/job:worker/task:{}".format(FLAGS.task_index), cluster=cluster_spec)):
  154. images, labels = inputs(FLAGS.batch_size)
  155. logits = inference(images, FLAGS.hidden1, FLAGS.hidden2)
  156. loss = lossFunction(logits, labels)
  157. train_op = training(loss, FLAGS.learning_rate)
  158. with tf.train.MonitoredTrainingSession(
  159. master=server.target,
  160. is_chief=(FLAGS.task_index == 0),
  161. checkpoint_dir=FLAGS.train_dir) as sess:
  162. while not sess.should_stop():
  163. sess.run(train_op)
  164.  
  165. if __name__ == "__main__":
  166. tf.app.run()

四、分布式模型的啟動

首先關閉防火牆

sudo iptable –F

然后在不同的機器上面啟動服務

  1. #在246.1機器上面運行參數服務器，命令：
  2. CLASSPATH=$($HADOOP_HDFS_HOME/bin/hadoop classpath --glob) python /home/bdusr01/tine/Distributed_Tensorflow_MNIST_Model_Used_NN_Read_TFRecords_On_HDFS_Support_Kerberos.py --ps_hosts=10.142.246.1:1120 --worker_hosts=10.142.78.41:1121,10.142.78.45:1122 --data_dir=hdfs://default/user/bdusr01/asy/MNIST_data --train_dir=/home/bdusr01/checkpoint/ --job_name=ps --task_index=0
  3.  
  4.  
  5. #在78.41機器上面運行worker0，命令：
  6. CLASSPATH=$($HADOOP_HDFS_HOME/bin/hadoop classpath --glob) python /home/bdusr01/tine/Distributed_Tensorflow_MNIST_Model_Used_NN_Read_TFRecords_On_HDFS_Support_Kerberos.py --ps_hosts=10.142.246.1:1120 --worker_hosts=10.142.78.41:1121,10.142.78.45:1122 --data_dir=hdfs://default/user/bdusr01/asy/MNIST_data --train_dir=/home/bdusr01/checkpoint/ --job_name=worker --task_index=0
  7.  
  8. #在78.45機器上面運行worker1，命令：
  9. CLASSPATH=$($HADOOP_HDFS_HOME/bin/hadoop classpath --glob) python /home/bdusr01/tine/Distributed_Tensorflow_MNIST_Model_Used_NN_Read_TFRecords_On_HDFS_Support_Kerberos.py--ps_hosts=10.142.246.1:1120 --worker_hosts=10.142.78.41:1121,10.142.78.45:1122 --data_dir=hdfs://default/user/bdusr01/asy/MNIST_data --train_dir=/home/bdusr01/checkpoint/ --job_name=worker --task_index=1
  10.  
  11. #在78.41機器上面運行監控，命令：
  12. tensorboard --logdir=/home/bdusr01/checkpoint/

五、模型監控

我們在剛剛的41機器上面啟動了TensorBoard，可以通過地址http://10.142.78.41:6006/進行模型的監控。模型訓練過程中參數可以動態的進行觀測，示例如下：

圖片描述

模型的網絡結構可以詳細的參看每個細節，示例如下：

圖片描述

當我們利用分布式的Tensorflow對大數據進行訓練完成后，可以利用Bazel構建一個靈活高可用的服務–TensorFlow Serving，能夠很方便的將深度學習生產化，解決了模型無法提供服務的弊端。到此為止，本文就將自己項目中的一個基礎模塊的示例介紹完了，本項目更有含金量的是模型建立、工程開發、業務邏輯部分，如有機會再進行更詳細的交流作者：丁廷鶴，碩士期間在復旦大學計算機學院上海市智能信息重點實驗室從事數據挖掘學習，目前在上海一家央企總部工作，從事大數據領域spark全棧模塊、機器學習、深度學習方面的開發和研究。


免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。



猜您在找 遙感大數據深度學習變化檢測 大數據學習腦圖以及7個入門教程！ 大數據學習之四——HDFS 大數據技術之_03_Hadoop學習_01_入門_大數據概論+從Hadoop框架討論大數據生態+Hadoop運行環境搭建（開發重點） 【大數據】科普一下大數據的那些事兒 大數據開源框架技術匯總 大數據 常用框架特點 大數據視頻教程合集 CRL快速開發框架系列教程十一(大數據分庫分表解決方案)
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM