for r in range(RNNS):    with tf.variable_scope(''recurent_%d'' % r) as scope:        state = [tf.zeros((BATCH_SIZE, sz)) for sz in rnn_func.state_size]        time_outputs = [None] * TIME_STEPS        for t in range(TIME_STEPS):            rnn_input = getTimeStep(rnn_outputs[r - 1], t)            time_outputs[t], state = rnn_func(rnn_input, state)            time_outputs[t] = tf.reshape(time_outputs[t], (-1, 1, RNN_SIZE))            scope.reuse_variables()        rnn_outputs[r] = tf.concat(1, time_outputs)

for r in range(RNNS):    with tf.variable_scope(''recurent_%d'' % r) as scope:        saved_state = tf.get_variable(''saved_state'', ...)        rnn_outputs[r], state = rnn_func(rnn_outputs[r - 1], saved_state)        saved_state = tf.assign(saved_state, state)

def get_state_variables(batch_size, cell):    # For each layer, get the initial state and make a variable out of it    # to enable updating its value.    state_variables = []    for state_c, state_h in cell.zero_state(batch_size, tf.float32):        state_variables.append(tf.contrib.rnn.LSTMStateTuple(            tf.Variable(state_c, trainable=False),            tf.Variable(state_h, trainable=False)))    # Return as a tuple, so that it can be fed to dynamic_rnn as an initial state    return tuple(state_variables)def get_state_update_op(state_variables, new_states):    # Add an operation to update the train states with the last state tensors    update_ops = []    for state_variable, new_state in zip(state_variables, new_states):        # Assign the new state to the state variables on this layer        update_ops.extend([state_variable[0].assign(new_state[0]),                           state_variable[1].assign(new_state[1])])    # Return a tuple in order to combine all update_ops into a single operation.    # The tuple''s actual value should not be used.    return tf.tuple(update_ops)

data = tf.placeholder(tf.float32, (batch_size, max_length, frame_size))cell_layer = tf.contrib.rnn.GRUCell(256)cell = tf.contrib.rnn.MultiRNNCell([cell] * num_layers)# For each layer, get the initial state. states will be a tuple of LSTMStateTuples.states = get_state_variables(batch_size, cell)# Unroll the LSTMoutputs, new_states = tf.nn.dynamic_rnn(cell, data, initial_state=states)# Add an operation to update the train states with the last state tensors.update_op = get_state_update_op(states, new_states)sess = tf.Session()sess.run(tf.global_variables_initializer())sess.run([outputs, update_op], {data: ...})

def get_state_reset_op(state_variables, cell, batch_size):    # Return an operation to set each variable in a list of LSTMStateTuples to zero    zero_states = cell.zero_state(batch_size, tf.float32)    return get_state_update_op(state_variables, zero_states)

reset_state_op = get_state_reset_op(state, cell, max_batch_size)# Reset the state to zero before feeding inputsess.run([reset_state_op])sess.run([outputs, update_op], {data: ...})

在Centos6系统上安装Hadoop、Spark集群，并使用TensorFlowOnSpark的 YARN运行模式下执行TensorFlow的代码。（最好可以在不联网的集群中进行配置并运行）

2. 系统环境（拓扑）

操作系统：Centos6.5 Final ； Hadoop：2.7.4 ； Spark：1.5.1-Hadoop2.6； TensorFlow 1.3.0；TensorFlowOnSpark （github最新下载）；Python：2.7.12；

s0.centos.com: memory:1.5G namenode/resourcemanager ; 1核

s1.centos.com / s2.centos.com/ s3.centos.com : datanode/nodemanager ; memory： 1.2G， 1 核

其中yarn-site.xml 部分配置如下（参考默认的，TensorFlowonspark运行不起来）：

<property>
                <name>yarn.scheduler.maximum-allocation-mb</name>
                <value>2048</value>
            </property>
        <property>
                <name>yarn.nodemanager.resource.memory-mb</name>
                <value>2048</value>
        </property>
        <property>
                <name>yarn.nodemanager.resource.cpu-vcores</name>
                        <value>2</value>
                            </property>

3. 参考

https://blog.abysm.org/2016/06/building-tensorflow-centos-6/： Centos6 build TensorFlow

TensorFlow github wiki ：https://github.com/yahoo/TensorFlowOnSpark/wiki/GetStarted_YARN ; installTensorFlowOnSpark ；

TensorFlow github wiki： https://github.com/yahoo/TensorFlowOnSpark/wiki/Conversion-Guide ;conversionTensorFlow code ；

步骤如下：

详细步骤如下：

1.安装devtoolset-6 及Python：

安装repo库： yum install -y centos-release-scl
安装 devtoolset:  yum install -y devtoolset-6 

安装Python：

yum install python27 python27-numpy python27-python-devel python27-python-wheel

安装一些常用包：

yum install –y vim zip unzip openssh-clients

2.下载bazel，这里下载的是0.5.1（虽然也下载了0.4.X的版本，下载包难下）

先执行：
export CC=/opt/rh/devtoolset-6/root/usr/bin/gcc
接着进入编译环境：
scl enable devtoolset-6 python27 bash
接着以此执行：
 unzip bazel-0.5.1-dist.zip -d bazel-0.5.1-dist
cd bazel-0.5.1-dist

# compile
./compile.sh
 
# install
mkdir -p ~/bin
cp output/bazel ~/bin/

exit  //退出scl环境
// 耗时较久

3.下载TensorFlow1.3.0源码并解压

4.进入tensorflow-1.3.0 ，修改tensorflow/tensorflow.bzl文件中的tf_extension_linkopts函数如下形式：(添加一个-lrt)

def tf_extension_linkopts():
  return ["-lrt"]  # No extension link opts

5.编译安装TensorFlow：

安装基本软件： yum install –y patch
接着，进入编译环境：
scl enable devtoolset-6 python27 bash
cd tensorflow-1.3.0
./configure
 
# build
~/bin/bazel build --config=opt //tensorflow/tools/pip_package:build_pip_package
bazel-bin/tensorflow/tools/pip_package/build_pip_package /tmp/tensorflow_pkg
exit // 退出编译环境
// 耗时同样很久，同样使用bazel0.4.X的版本编译TensorFlow1.3提示版本过低

编译后在/tmp/tensorflow_pkg则会生成一个TensorFlow的 安装包 ，并且是属于当前系统也就是Centos系统的安装包；

http://download.csdn.net/download/fansy1990/10042475 <<--- whl安装包下载地址

由于不想让现有的系统过于复杂，也就是直接在每个节点安装Python，然后安装TensorFlow等相关 Python包，所以参考TensorFlow on spark 官网进行，如下步骤：

6.安装Python自定义包(保持在联网状态下)；

由于想在未联网的情况下使用TensorFlow以及TensorFlowOnSpark，所以参考TensorFlowOnSpark github WIKI，直接编译一个Python包，并且把TensorFlow、TensorFlowOnSpark及其他常用module安装在这个Python包中，后面就可以直接把这个包上传到HDFS，使得各个子节点都可以共享共同一个Python.zip包的环境变量。

export PYTHON_ROOT=~/Python // 设置环境变量，并下载Python
curl -O https://www.python.org/ftp/python/2.7.12/Python-2.7.12.tgz
tar -xvf Python-2.7.12.tgz

编译并安装Python：

pushd Python-2.7.12
./configure --prefix="${PYTHON_ROOT}" --enable-unicode=ucs4
make
make install
popd

安装Pip：

pushd "${PYTHON_ROOT}"
curl -O https://bootstrap.pypa.io/get-pip.py
bin/python get-pip.py
popd

安装TensorFlow：

pushd "${PYTHON_ROOT}"
bin/pip install /tmp/tensorflow_pkg/tensorflow-1.3.0-cp27-none-linux_x86_64.whl
popd

在安装TensorFlow的时候会自动安装诸如 numpy等常用Python包；

安装TensorFlowOnSpark：

pushd "${PYTHON_ROOT}"
bin/pip install tensorflowonspark
popd

把“武装”好的Python打包并上传到HDFS：

pushd "${PYTHON_ROOT}"
zip -r Python.zip *
popd

hadoop fs -put ${PYTHON_ROOT}/Python.zip

现在就可以使用TensorFlow了；

7. 修改TensorFlow代码，比如下面的TensorFlow代码是可以在TensorFlow环境中运行的：

# from __future__ import absolute_import
# from __future__ import division
# from __future__ import print_function

import numpy as np

import tensorflow as tf

X_FEATURE = 'x'  # Name of the input feature.

train_percent = 0.8


def load_data(data_file_name):
    data = np.loadtxt(open(data_file_name),delimiter=",",skiprows=0)
    return data


def data_selection(iris,train_per):
    data,target = np.hsplit(iris[np.random.permutation(iris.shape[0])],np.array([-1]))

    row_split_index = int(data.shape[0] * train_per)

    x_train,x_test = (data[1:row_split_index],data[row_split_index:])
    y_train,y_test = (target[1:row_split_index],target[row_split_index:])
    return x_train,x_test,y_train.astype(int),y_test.astype(int)


def run():
    # Load dataset.
    data_file = 'iris01.csv'
    iris = load_data(data_file)
    # x_train,y_train,y_test = model_selection.train_test_split(
    #     iris.data,iris.target,test_size=0.2,random_state=42)

    x_train,y_test = data_selection(iris,train_percent)

    # print(x_test)
    # print(y_test)

    #
    # # Build 3 layer DNN with 10,20,10 units respectively.
    feature_columns = [
        tf.feature_column.numeric_column(
            X_FEATURE,shape=np.array(x_train).shape[1:])]
    classifier = tf.estimator.DNNClassifier(
        feature_columns=feature_columns,hidden_units=[10,10],n_classes=3)
    #
    # # Train.
    train_input_fn = tf.estimator.inputs.numpy_input_fn(
        x={X_FEATURE: x_train},y=y_train,num_epochs=None,shuffle=True)
    classifier.train(input_fn=train_input_fn,steps=200)
    #
    # # Predict.
    test_input_fn = tf.estimator.inputs.numpy_input_fn(
        x={X_FEATURE: x_test},y=y_test,num_epochs=1,shuffle=False)
    predictions = classifier.predict(input_fn=test_input_fn)
    y_predicted = np.array(list(p['class_ids'] for p in predictions))
    y_predicted = y_predicted.reshape(np.array(y_test).shape)
    # #
    # # # score with sklearn.
    # score = metrics.accuracy_score(y_test,y_predicted)
    # print('Accuracy (sklearn): {0:f}'.format(score))
    print(np.concatenate(( y_predicted,y_test),axis= 1))
    # score with tensorflow.
    scores = classifier.evaluate(input_fn=test_input_fn)
    print('Accuracy (tensorflow): {0:f}'.format(scores['accuracy']))

    print(classifier.params)


if __name__ == '__main__':
    run()

其中iris01.csv 数据如下：

5.1,3.5,1.4,0.2,0
4.9,3.0,0
4.7,3.2,1.3,0
4.6,3.1,1.5,0
5.0,3.6,0
5.4,3.9,1.7,0.4,3.4,0.3,0
4.4,2.9,0.1,3.7,0
4.8,1.6,0
4.3,1.1,0
5.8,4.0,1.2,0
5.7,4.4,0
5.1,3.8,1.0,3.3,0.5,1.9,0
5.2,4.1,0
5.5,4.2,0
4.5,2.3,0.6,0
5.3,0
7.0,4.7,1
6.4,4.5,1
6.9,4.9,1
5.5,1
6.5,2.8,4.6,1
5.7,1
6.3,1
4.9,2.4,1
6.6,1
5.2,2.7,1
5.0,2.0,1
5.9,1
6.0,2.2,1
6.1,1
5.6,1
6.7,1
5.8,1
6.2,2.5,4.8,1.8,4.3,1
6.8,5.0,2.6,5.1,1
5.4,1
5.1,6.0,2
5.8,2
7.1,5.9,2.1,2
6.3,5.6,2
6.5,5.8,2
7.6,6.6,2
4.9,2
7.3,6.3,2
6.7,2
7.2,6.1,2
6.4,5.3,2
6.8,5.5,2
5.7,2
7.7,6.7,6.9,2
6.0,2
6.9,5.7,2
5.6,2
6.2,2
6.1,2
7.4,2
7.9,6.4,5.4,5.2,2
5.9,2

那代码怎么修改呢？

1）. 导入必要的包：

from pyspark.context import SparkContext
from pyspark.conf import SparkConf
from tensorflowonspark import TFCluster,TFNode
#from com.yahoo.ml.tf import TFCluster,TFNode
from datetime import datetime

这里要注意，导入TFCluster的时候，不要参考官网的导入方式，而应该从tensorflowonspark导入；

2.） 修改main函数，比如我这里的函数run，只需要添加两个参数即可：（argv,cxt）

3) 把原来的main函数调用，替换成下面的调用方式 ，比如我这里原来只需要在main函数执行run即可，这里需要调用TFCluster.run，并且把我的run函数传递给第二个参数值：

sc = SparkContext(conf=SparkConf().setAppName("your_app_name"))
    num_executors = int(sc._conf.get("spark.executor.instances"))
    num_ps = 1
    tensorboard = True

    cluster = TFCluster.run(sc,run,sys.argv,num_executors,num_ps,tensorboard,TFCluster.InputMode.TENSORFLOW)
    cluster.shutdown()

然后就可以运行了，修改后的代码如下：

# from __future__ import absolute_import
# from __future__ import division
# from __future__ import print_function
from pyspark.context import SparkContext
from pyspark.conf import SparkConf
from tensorflowonspark import TFCluster,TFNode
from datetime import datetime
import numpy as np
import sys
# from sklearn import metrics
# from sklearn import model_selection

import tensorflow as tf

X_FEATURE = 'x'  # Name of the input feature.

train_percent = 0.8


def load_data(data_file_name):
    data = np.loadtxt(open(data_file_name),y_test.astype(int)


def map_run(argv,ctx):
    # Load dataset.
    data_file = 'iris01.csv'
    iris = load_data(data_file)
    # x_train,axis= 1))
    # score with tensorflow.
    scores = classifier.evaluate(input_fn=test_input_fn)
    print('Accuracy (tensorflow): {0:f}'.format(scores['accuracy']))

    print(classifier.params)


if __name__ == '__main__':
    import tensorflow as tf
    import sys
    sc = SparkContext(conf=SparkConf().setAppName("your_app_name"))
    num_executors = int(sc._conf.get("spark.executor.instances"))
    num_ps = 1
    tensorboard = False

    cluster = TFCluster.run(sc,map_run,TFCluster.InputMode.TENSORFLOW)
    cluster.shutdown()	

7. 设置环境变量，并运行：

1)上传iris01.csv到HDFS： hdfs dfs -put iris01.csv

2) 设置环境变量：

export PYTHON_ROOT=./Python
export LD_LIBRARY_PATH=${PATH}
export PYSPARK_PYTHON=${PYTHON_ROOT}/bin/python
export SPARK_YARN_USER_ENV="PYSPARK_PYTHON=Python/bin/python"
export PATH=${PYTHON_ROOT}/bin/:$PATH
#export QUEUE=gpu

# set paths to libjvm.so,libhdfs.so,and libcuda*.so
#export LIB_HDFS=/opt/cloudera/parcels/CDH/lib64                      # for CDH (per @wangyum)
export LIB_HDFS=$HADOOP_PREFIX/lib/native
export LIB_JVM=$JAVA_HOME/jre/lib/amd64/server
#export LIB_CUDA=/usr/local/cuda-7.5/lib64

# for cpu mode:
 export QUEUE=default

3） 调用代码：

/usr/local/spark-1.5.1-bin-hadoop2.6/bin/spark-submit --master yarn --deploy-mode cluster --num-executors 3 --executor-memory 1024m --archives hdfs://s0:8020/user/root/Python.zip#Python,/root/iris01.csv /root/iris_c.py

4） 查看yarn日志，可以看到执行成功；

1） libc.so.6: version `GLIBC_2.14' not found

这个问题是由于Centos6的版本其GLIBC的版本是2.12 ，版本过低导致的；

解决思路：

a. 升级版本， 这个选项不适用，由于这个软件是底层软件，升级后导致系统不稳定；

b. 编译一个可以在Centos6上运行的TensorFlow安装包，也就是本文的做法；

2）Cannot run program "patch" (in directory "/root/.cache/bazel/_bazel_root/6093305914d4a581ed00c0f6c06f975b/external/boringssl")

yum install patch

3)Traceback (most recent call last):
File "iris_c.py",line 6,in <module>
from com.yahoo.ml.tf import TFCluster,TFNode
ImportError: No module named com.yahoo.ml.tf

修改：
from com.yahoo.ml.tf import TFCluster,TFNode
=》
from tensorflowonspark import TFCluster,TFNode

1. 在编译tensorflow的时候遇到很多问题，使用bing的国际版查询效果会更好；

2. 暂时只能使用终端设置环境变量的方式执行程序，并且程序执行很慢，后面可以考虑使用开发工具直连提交任务，并着手提升效率；

分享，成长，快乐 脚踏实地，专注 转载请注明blog地址：http://blog.csdn.net/fansy1990

import tensorflow as tf

if __name__==''__main__'': 
    g = tf.Graph()   
    # add ops to the user created graph
    with g.as_default(): 
        hello = tf.constant(''Hello Tensorflow'') 
        sess = tf.compat.v1.Session(graph=g)
        print(sess.run(hello)) 

warmup_steps = 50
def loss_mse_warmup(y_true,y_pred):
    """
    Calculate the Mean Squared Error between y_true and y_pred,but ignore the beginning "warmup" part of the sequences.
    
    y_true is the desired output.
    y_pred is the model''s output.
    """

    # The shape of both input tensors are:
    # [batch_size,sequence_length,num_y_signals].

    # Ignore the "warmup" parts of the sequences
    # by taking slices of the tensors.
    y_true_slice = y_true[:,warmup_steps:,:]
    y_pred_slice = y_pred[:,:]

    # These sliced tensors both have this shape:
    # [batch_size,sequence_length - warmup_steps,num_y_signals]

    # Calculat the Mean Squared Error and use it as loss.
    mse = mean(square(y_true_slice - y_pred_slice))
    
    return mse

model = Sequential()

model.add(GRU(units=128,return_sequences=True,input_shape=(None,num_x_signals)))

model.add(Dense(num_y_signals,activation=''sigmoid'')) 

optimizer = Adam(clipvalue=1)

model.compile(loss=loss_mse_warmup,optimizer=optimizer)

```python
import tensorflow as tf
from tensorflow.contrib import rnn
from tensorflow.examples.tutorials.mnist import  input_data
mnist=input_data.read_data_sets("MNNIST_data",one_hot=True)

#输入图片为 28*28
n_inputs=28#输入一行，一行有28个像素
max_time=28#一共28行，所以为28*28
lstm_size=100#100个隐藏单元
batch_size=50
n_classes=10
n_batch=mnist.train.num_examples//batch_size#计算一共多少批次

#这里none表示第一个维度可以是任意长度
x=tf.placeholder(tf.float32,[None,784])

y=tf.placeholder(tf.float32,[None,10])

#初始化权值
weights=tf.Variable(tf.truncated_normal([lstm_size,n_classes],stddev=0.1))
#初始化偏置值
biases=tf.Variable(tf.constant(0.1,shape=[n_classes]))

##定义Rnn 网络
def RNN(X,weights,biases):
    inputs=tf.reshape(X,[-1,max_time,n_inputs])
    #定义lstm基本cell
    lstm_cell = rnn.BasicLSTMCell(lstm_size)
    #lstm_cell=tf.contrib.rnn.core_rnn_cell.BasicLSTMCell(lstm_size)
    outputs,final_state=tf.nn.dynamic_rnn(lstm_cell,inputs,dtype=tf.float32)
    results=tf.nn.softmax(tf.matmul(final_state[1],weights)+biases)
    return results
prediction=RNN(x,weights,biases)
#损失函数
cross_entropy=tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits(logits=prediction,labels=y))
#优化器
train_step=tf.train.AdamOptimizer(1e-4).minimize(cross_entropy)
#保存结果
correct_prediction=tf.equal(tf.argmax(y,1),tf.argmax(prediction,1))

accuracy=tf.reduce_mean(tf.cast(correct_prediction,tf.float32))

init=tf.global_variables_initializer()

with tf.Session() as sess:
    sess.run(init)
    for epoch in range(6):
        for batch in range(n_batch):
            batch_xs,batch_ys=mnist.train.next_batch(batch_size)
            sess.run(train_step,feed_dict={x:batch_xs,y:batch_ys})

        acc=sess.run(accuracy,feed_dict={x:mnist.test.images,y:mnist.test.labels})
        print("iter:"+str(epoch)+"testing accuracy"+str(acc))