PAI-TF提供的TableRecordDataset與原生TensorFlow RecordDataset相似，可以為數(shù)據(jù)變換（Transformation）的Dataset接口提供數(shù)據(jù)源。TableRecordDataset的接口定義如下。

class TableRecordDataset(Dataset):
  def __init__(self,
               filenames,
               record_defaults,
               selected_cols=None,
               excluded_cols=None,
                slice_id=0,
                slice_count=1,
               num_threads=0,
               capacity=0):

TableRecordDataset返回一個新的Dataset對象，可以作為Pipeline工作流構(gòu)建的輸入。

### other definition codes was ignored here.

# Suppose an odps table named 'sample_table' was built in
# 'test' project, which includes 5 columns:
#   (itemid bigint, name string, price double,
#    virtual bool, tags string)

# Table name would be passed from run commands.
tables = ["odps://test/tables/sample_table"]

# Firstly, we define a new TableRecordDataset to read itemid and price.
dataset = tf.data.TableRecordDataset(tables,
                                     record_defaults = (0, 0.0),
                                     selected_cols = "itemid, price")
# Get a batch of 128
dataset = dataset.batch(128)
# Set epoch as 10
dataset = dataset.repeat(10)
# At last we got a batch of ids and prices.
[ids, prices] = dataset.make_one_shot_iterator().get_next()

### Then we do other graph construction and finally run the graph.

執(zhí)行Session時調(diào)用get_next()方法，從表中讀取128行數(shù)據(jù)，并根據(jù)record_defaults指定的類型將每列數(shù)據(jù)解析為對應(yīng)類型的Tensor。其中get_next()返回的output_types需要與record_defaults的參數(shù)類型相同，output_shapes的Tensor Shape需要與record_defaults的元素數(shù)量一致。

以邏輯回歸（Logistic Regression）為例，介紹如何使用TableRecordDataset讀取表數(shù)據(jù)并進行模型訓(xùn)練。

1. 數(shù)據(jù)準備。

   TableRecordReader是將整行數(shù)據(jù)作為一個字符串導(dǎo)入MaxCompute表，讀取之后再進行解析。而使用TableRecordDataset時，建議MaxCompute數(shù)據(jù)表按照列存放相應(yīng)的數(shù)據(jù)，Dataset接口會將表中的數(shù)據(jù)以指定類型的Tensor返回。

   1. 創(chuàng)建表。

      使用MaxCompute創(chuàng)建一個包含四列數(shù)據(jù)的表。

      odps@ algo_platform_dev>create table sample (col1 double, col2 double, col3 double, col4 double);
      Data Health Manager:Your health synthesize score is 5, so, your job priority is 7
      
      ID = 201803050245351****6Vgsxo2
      OK
      odps@ algo_platform_dev>read sample;
      +------------+------------+------------+------------+
      | col1       | col2       | col3       | col4       |
      +------------+------------+------------+------------+
      +------------+------------+------------+------------+

   2. 導(dǎo)入數(shù)據(jù)。

      下載測試數(shù)據(jù)，并使用MaxCompute Console Tunnel命令將其導(dǎo)入MaxCompute表。

      #查看下載的測試數(shù)據(jù)。
      $head -n 3 sample.csv
      0,0,0.017179100152531324,1
      0,1,0.823381420409002,1
      0,2,1.6488850495540865,1

      #將數(shù)據(jù)導(dǎo)入MaxCompute表。
      odps@ algo_platform_dev>tunnel upload /tmp/data/sample.csv sample -fd=,;
      Upload session: 20180305135640c8cc650a0000****
      Start upload:sample.csv
      Using \n to split records
      Upload in strict schema mode: true
      Total bytes:260093   Split input to 1 blocks
      2018-03-05 13:56:40 scan block: '1'
      2018-03-05 13:56:40 scan block complete, blockid=1
      2018-03-05 13:56:40 upload block: '1'
      2018-03-05 13:56:41 upload block complete, blockid=1
      upload complete, average speed is 254 KB/s
      OK
      odps@ algo_platform_dev>read sample 3;
      +------------+------------+------------+------------+
      | col1       | col2       | col3       | col4       |
      +------------+------------+------------+------------+
      | 0.0        | 0.0        | 0.017179100152531324 | 1.0        |
      | 0.0        | 1.0        | 0.823381420409002 | 1.0        |
      | 0.0        | 2.0        | 1.6488850495540865 | 1.0        |
      +------------+------------+------------+------------+

      說明

      因為該測試數(shù)據(jù)的每行內(nèi)容使用英文逗號（,）分隔，所以使用-fd=,配置分隔符為英文逗號（,）才能將每行數(shù)據(jù)分為四列導(dǎo)入至相應(yīng)的MaxCompute表。

2. 構(gòu)建輸入數(shù)據(jù)和模型。

   構(gòu)建輸入數(shù)據(jù)的示例代碼如下。除無需定義tf.train.Coordinator和運行start_queue_runners以外，其余代碼與使用TableRecordReader的代碼相同。

   #define the input
   def input_fn():
       dataset = tf.data.TableRecordDataset([FLAGS.tables], record_defaults=[1.0]*4).repeat().batch(128)
       v1, v2, v3, v4 = dataset.make_one_shot_iterator().get_next()
       labels = tf.reshape(tf.cast(v4, tf.int32), [128])
       features = tf.stack([v1, v2, v3], axis=1)
       return features, labels

   完整的示例代碼lr_dataset.py如下。

   import tensorflow as tf
   
   tf.app.flags.DEFINE_string("tables", "", "tables info")
   FLAGS = tf.app.flags.FLAGS
   
   #define the input
   def input_fn():
       dataset = tf.data.TableRecordDataset([FLAGS.tables], record_defaults=[1.0]*4).repeat().batch(128)
       v1, v2, v3, v4 = dataset.make_one_shot_iterator().get_next()
       labels = tf.reshape(tf.cast(v4, tf.int32), [128])
       features = tf.stack([v1, v2, v3], axis=1)
       return features, labels
   
   #construct the model
   def model_fn(features, labels):
       W = tf.Variable(tf.zeros([3, 2]))
       b = tf.Variable(tf.zeros([2]))
       pred = tf.matmul(features, W) + b
   
       loss =  tf.reduce_mean(tf.nn.sparse_softmax_cross_entropy_with_logits(logits=pred,labels=labels))
   
       # Gradient Descent
       optimizer = tf.train.GradientDescentOptimizer(0.05).minimize(loss)
       return loss, optimizer
   
   features, labels = input_fn()
   loss, optimizer = model_fn(features, labels)
   
   init = tf.global_variables_initializer()
   local_init = tf.local_variables_initializer()
   
   sess = tf.Session()
   sess.run(init)
   sess.run(local_init)
   
   for step in range(10000):
       _, c = sess.run([optimizer, loss])
       if step % 2000 == 0:
           print("loss," , c)

3. 提交任務(wù)。

   odps@ algo_platform_dev>pai -name tensorflow1120_cpu_ext -Dtables=odps://algo_platform_dev/tables/sample -Dscript=file:///tmp/lr_dataset.py;

4. 查看執(zhí)行結(jié)果。

   單擊提交任務(wù)返回的Logview鏈接，查看執(zhí)行結(jié)果。

   start launching tensorflow job
   ('loss,', 0.6931472)
   ('loss,', 0.007929571)
   ('loss,', 0.0016527221)
   ('loss,', 0.0023481336)
   ('loss,', 0.0011788738)