pai_online_project.iris_data

PAI -name tensorflow180_ext -project algo_public -Doutputs="odps://${当前项目名}/tables/${输出表名}" -DossHost="${OSS的host}" -Dtables="odps://${当前项目名}/tables/${输入表名}" -DgpuRequired="${GPU卡数}" -Darn="${OSS访问RoleARN}" -Dscript="${执行的代码文件}";

import tensorflow as tf

tf.app.flags.DEFINE_string("tables", "", "tables info")

FLAGS = tf.app.flags.FLAGS

print("tables:" + FLAGS.tables)

tables = [FLAGS.tables]

filename_queue = tf.train.string_input_producer(tables, num_epochs=1)

reader = tf.TableRecordReader()

key, value = reader.read(filename_queue)

record_defaults = [[1.0], [1.0], [1.0], [1.0], ["Iris-virginica"]]

col1, col2, col3, col4, col5 = tf.decode_csv(value, record_defaults = record_defaults)

# line 9 and 10 can be written like below for short. It will be helpful when too many columns exist.

# record_defaults = [[1.0]] * 4 + [["Iris-virginica"]]

# value_list = tf.decode_csv(value, record_defaults = record_defaults)

writer = tf.TableRecordWriter("odps://pai_bj_test2/tables/iris_output")

write_to_table = writer.write([0, 1, 2, 3, 4], [col1, col2, col3, col4, col5])

# line 16 can be written like below for short. It will be helpful when too many columns exist.

# write_to_table = writer.write(range(5), value_list)

close_table = writer.close()

init = tf.global_variables_initializer()

with tf.Session() as sess:

    sess.run(init)

    sess.run(tf.local_variables_initializer())

    coord = tf.train.Coordinator()

    threads = tf.train.start_queue_runners(coord=coord)

    try:

       step = 0

       while not coord.should_stop():

            step += 1

            sess.run(write_to_table)

    except tf.errors.OutOfRangeError:

        print('%d records copied' % step)

    finally:

        sess.run(close_table)

        coord.request_stop()

        coord.join(threads)