如何*实际*读取 TensorFlow 中的 CSV 数据？
How to *actually* read CSV data in TensorFlow?
我对 TensorFlow 的世界还比较陌生，对你如何实际上 将 CSV 数据读入TensorFlow 中可用的 example/label 个张量。 TensorFlow tutorial on reading CSV data 中的示例非常零散，只能让您了解如何使用 CSV 数据进行训练。
这是我根据 CSV 教程拼凑而成的代码：
from __future__ import print_function
import tensorflow as tf
def file_len(fname):
with open(fname) as f:
for i, l in enumerate(f):
pass
return i + 1
filename = "csv_test_data.csv"
# setup text reader
file_length = file_len(filename)
filename_queue = tf.train.string_input_producer([filename])
reader = tf.TextLineReader(skip_header_lines=1)
_, csv_row = reader.read(filename_queue)
# setup CSV decoding
record_defaults = [[0],[0],[0],[0],[0]]
col1,col2,col3,col4,col5 = tf.decode_csv(csv_row, record_defaults=record_defaults)
# turn features back into a tensor
features = tf.stack([col1,col2,col3,col4])
print("loading, " + str(file_length) + " line(s)\n")
with tf.Session() as sess:
tf.initialize_all_variables().run()
# start populating filename queue
coord = tf.train.Coordinator()
threads = tf.train.start_queue_runners(coord=coord)
for i in range(file_length):
# retrieve a single instance
example, label = sess.run([features, col5])
print(example, label)
coord.request_stop()
coord.join(threads)
print("\ndone loading")
这是我正在加载的 CSV 文件中的一个简短示例 - 非常基本的数据 - 4 个特征列和 1 个标签列：
0,0,0,0,0
0,15,0,0,0
0,30,0,0,0
0,45,0,0,0
上面的所有代码都是从 CSV 文件中一个一个地打印每个示例，虽然不错，但对于训练来说非常无用。
我在这里纠结的是你如何将这些单独的例子逐个加载到训练数据集中。例如，here's a notebook 我在 Udacity 深度学习课程中学习。我基本上想获取正在加载的 CSV 数据，然后将其放入 train_dataset 和 train_labels 之类的内容中:
def reformat(dataset, labels):
dataset = dataset.reshape((-1, image_size * image_size)).astype(np.float32)
# Map 2 to [0.0, 1.0, 0.0 ...], 3 to [0.0, 0.0, 1.0 ...]
labels = (np.arange(num_labels) == labels[:,None]).astype(np.float32)
return dataset, labels
train_dataset, train_labels = reformat(train_dataset, train_labels)
valid_dataset, valid_labels = reformat(valid_dataset, valid_labels)
test_dataset, test_labels = reformat(test_dataset, test_labels)
print('Training set', train_dataset.shape, train_labels.shape)
print('Validation set', valid_dataset.shape, valid_labels.shape)
print('Test set', test_dataset.shape, test_labels.shape)
我试过像这样使用 tf.train.shuffle_batch，但它莫名其妙地挂起：
for i in range(file_length):
# retrieve a single instance
example, label = sess.run([features, colRelevant])
example_batch, label_batch = tf.train.shuffle_batch([example, label], batch_size=file_length, capacity=file_length, min_after_dequeue=10000)
print(example, label)
总而言之，这是我的问题：
- 这个过程我错过了什么？
- 感觉我缺少一些关于如何正确构建输入管道的关键直觉。
- 有没有办法避免必须知道 CSV 文件的长度？
- 必须知道要处理的行数（上面的
for i in range(file_length) 行代码）感觉很不优雅
编辑：
一旦 Yaroslav 指出我可能在这里混淆了命令式和图形构造部分，它就开始变得更加清晰。我能够将以下代码整合在一起，我认为这更接近于从 CSV 训练模型时通常所做的（不包括任何模型训练代码）：
from __future__ import print_function
import numpy as np
import tensorflow as tf
import math as math
import argparse
parser = argparse.ArgumentParser()
parser.add_argument('dataset')
args = parser.parse_args()
def file_len(fname):
with open(fname) as f:
for i, l in enumerate(f):
pass
return i + 1
def read_from_csv(filename_queue):
reader = tf.TextLineReader(skip_header_lines=1)
_, csv_row = reader.read(filename_queue)
record_defaults = [[0],[0],[0],[0],[0]]
colHour,colQuarter,colAction,colUser,colLabel = tf.decode_csv(csv_row, record_defaults=record_defaults)
features = tf.stack([colHour,colQuarter,colAction,colUser])
label = tf.stack([colLabel])
return features, label
def input_pipeline(batch_size, num_epochs=None):
filename_queue = tf.train.string_input_producer([args.dataset], num_epochs=num_epochs, shuffle=True)
example, label = read_from_csv(filename_queue)
min_after_dequeue = 10000
capacity = min_after_dequeue + 3 * batch_size
example_batch, label_batch = tf.train.shuffle_batch(
[example, label], batch_size=batch_size, capacity=capacity,
min_after_dequeue=min_after_dequeue)
return example_batch, label_batch
file_length = file_len(args.dataset) - 1
examples, labels = input_pipeline(file_length, 1)
with tf.Session() as sess:
tf.initialize_all_variables().run()
# start populating filename queue
coord = tf.train.Coordinator()
threads = tf.train.start_queue_runners(coord=coord)
try:
while not coord.should_stop():
example_batch, label_batch = sess.run([examples, labels])
print(example_batch)
except tf.errors.OutOfRangeError:
print('Done training, epoch reached')
finally:
coord.request_stop()
coord.join(threads)
我认为您在这里混淆了命令式和图形构造部分。 tf.train.shuffle_batch操作创建了一个新的队列节点，单个节点可以用来处理整个数据集。所以我认为你挂了，因为你在 for 循环中创建了一堆 shuffle_batch 队列并且没有为它们启动队列 运行ners。
正常的输入管道用法如下所示：
- 将
shuffle_batch 等节点添加到输入管道
- （可选，以防止无意中修改图形）完成图形
---图构造结束，命令式编程开始--
tf.start_queue_runnerswhile(True): session.run()
为了更具可扩展性（避免 Python GIL），您可以使用 TensorFlow 管道生成所有数据。但是，如果性能不是很重要，您可以使用 slice_input_producer. 将 numpy 数组连接到输入管道这是一个包含一些 Print 节点的示例，以查看发生了什么（Print 中的消息当节点为 运行)
时转到标准输出
tf.reset_default_graph()
num_examples = 5
num_features = 2
data = np.reshape(np.arange(num_examples*num_features), (num_examples, num_features))
print data
(data_node,) = tf.slice_input_producer([tf.constant(data)], num_epochs=1, shuffle=False)
data_node_debug = tf.Print(data_node, [data_node], "Dequeueing from data_node ")
data_batch = tf.batch([data_node_debug], batch_size=2)
data_batch_debug = tf.Print(data_batch, [data_batch], "Dequeueing from data_batch ")
sess = tf.InteractiveSession()
sess.run(tf.initialize_all_variables())
tf.get_default_graph().finalize()
tf.start_queue_runners()
try:
while True:
print sess.run(data_batch_debug)
except tf.errors.OutOfRangeError as e:
print "No more inputs."
你应该看到这样的东西
[[0 1]
[2 3]
[4 5]
[6 7]
[8 9]]
[[0 1]
[2 3]]
[[4 5]
[6 7]]
No more inputs.
“8、9”号没有填满整批，所以没有生产。 tf.Print 也被打印到 sys.stdout，因此它们会单独显示在终端中。
PS: 最小连接 batch 到手动初始化的队列是在 github issue 2193
此外，出于调试目的，您可能需要在会话中设置 timeout，这样您的 IPython 笔记本就不会在空队列出队时挂起。我在我的会话中使用这个辅助函数
def create_session():
config = tf.ConfigProto(log_device_placement=True)
config.gpu_options.per_process_gpu_memory_fraction=0.3 # don't hog all vRAM
config.operation_timeout_in_ms=60000 # terminate on long hangs
# create interactive session to register a default session
sess = tf.InteractiveSession("", config=config)
return sess
可扩展性说明：
tf.constant 将您的数据副本内联到图表中。图定义的大小有 2GB 的基本限制，因此这是数据大小的上限 - 您可以通过使用
v=tf.Variable 并通过 运行 将 v.assign_op 与 tf.placeholder 放在右侧并将数据保存到其中来绕过该限制numpy 数组到占位符 (feed_dict)
- 这仍然会创建两个数据副本，因此为了节省内存，您可以制作自己的
slice_input_producer 版本，它在 numpy 数组上运行，并使用 feed_dict[= 一次上传一行60=]
或者你可以试试这个，代码使用 pandas 和 numpy 将 Iris 数据集加载到 tensorflow 中，并在会话中打印一个简单的神经元输出。希望对大家有个基本的了解有帮助.... [我还没有添加一个热解码标签的方式]。
import tensorflow as tf
import numpy
import pandas as pd
df=pd.read_csv('/home/nagarjun/Desktop/Iris.csv',usecols = [0,1,2,3,4],skiprows = [0],header=None)
d = df.values
l = pd.read_csv('/home/nagarjun/Desktop/Iris.csv',usecols = [5] ,header=None)
labels = l.values
data = numpy.float32(d)
labels = numpy.array(l,'str')
#print data, labels
#tensorflow
x = tf.placeholder(tf.float32,shape=(150,5))
x = data
w = tf.random_normal([100,150],mean=0.0, stddev=1.0, dtype=tf.float32)
y = tf.nn.softmax(tf.matmul(w,x))
with tf.Session() as sess:
print sess.run(y)
你可以使用最新的 tf.data API :
dataset = tf.contrib.data.make_csv_dataset(filepath)
iterator = dataset.make_initializable_iterator()
columns = iterator.get_next()
with tf.Session() as sess:
sess.run([iteator.initializer])
如果有人来这里寻找一种简单的方法来读取 tf.estimator API 中绝对大的和分片的 CSV 文件，那么请看下面我的代码
CSV_COLUMNS = ['ID','text','class']
LABEL_COLUMN = 'class'
DEFAULTS = [['x'],['no'],[0]] #Default values
def read_dataset(filename, mode, batch_size = 512):
def _input_fn(v_test=False):
# def decode_csv(value_column):
# columns = tf.decode_csv(value_column, record_defaults = DEFAULTS)
# features = dict(zip(CSV_COLUMNS, columns))
# label = features.pop(LABEL_COLUMN)
# return add_engineered(features), label
# Create list of files that match pattern
file_list = tf.gfile.Glob(filename)
# Create dataset from file list
#dataset = tf.data.TextLineDataset(file_list).map(decode_csv)
dataset = tf.contrib.data.make_csv_dataset(file_list,
batch_size=batch_size,
column_names=CSV_COLUMNS,
column_defaults=DEFAULTS,
label_name=LABEL_COLUMN)
if mode == tf.estimator.ModeKeys.TRAIN:
num_epochs = None # indefinitely
dataset = dataset.shuffle(buffer_size = 10 * batch_size)
else:
num_epochs = 1 # end-of-input after this
batch_features, batch_labels = dataset.make_one_shot_iterator().get_next()
#Begins - Uncomment for testing only -----------------------------------------------------<
if v_test == True:
with tf.Session() as sess:
print(sess.run(batch_features))
#End - Uncomment for testing only -----------------------------------------------------<
return add_engineered(batch_features), batch_labels
return _input_fn
TF.estimator中的用法示例：
train_spec = tf.estimator.TrainSpec(input_fn = read_dataset(
filename = train_file,
mode = tf.estimator.ModeKeys.TRAIN,
batch_size = 128),
max_steps = num_train_steps)
2.0 兼容解决方案：此答案可能由上述线程中的其他人提供，但我将提供有助于社区的其他链接。
dataset = tf.data.experimental.make_csv_dataset(
file_path,
batch_size=5, # Artificially small to make examples easier to show.
label_name=LABEL_COLUMN,
na_value="?",
num_epochs=1,
ignore_errors=True,
**kwargs)
更多信息，请参考此Tensorflow Tutorial。
我对 TensorFlow 的世界还比较陌生，对你如何实际上 将 CSV 数据读入TensorFlow 中可用的 example/label 个张量。 TensorFlow tutorial on reading CSV data 中的示例非常零散，只能让您了解如何使用 CSV 数据进行训练。
这是我根据 CSV 教程拼凑而成的代码：
from __future__ import print_function
import tensorflow as tf
def file_len(fname):
with open(fname) as f:
for i, l in enumerate(f):
pass
return i + 1
filename = "csv_test_data.csv"
# setup text reader
file_length = file_len(filename)
filename_queue = tf.train.string_input_producer([filename])
reader = tf.TextLineReader(skip_header_lines=1)
_, csv_row = reader.read(filename_queue)
# setup CSV decoding
record_defaults = [[0],[0],[0],[0],[0]]
col1,col2,col3,col4,col5 = tf.decode_csv(csv_row, record_defaults=record_defaults)
# turn features back into a tensor
features = tf.stack([col1,col2,col3,col4])
print("loading, " + str(file_length) + " line(s)\n")
with tf.Session() as sess:
tf.initialize_all_variables().run()
# start populating filename queue
coord = tf.train.Coordinator()
threads = tf.train.start_queue_runners(coord=coord)
for i in range(file_length):
# retrieve a single instance
example, label = sess.run([features, col5])
print(example, label)
coord.request_stop()
coord.join(threads)
print("\ndone loading")
这是我正在加载的 CSV 文件中的一个简短示例 - 非常基本的数据 - 4 个特征列和 1 个标签列：
0,0,0,0,0
0,15,0,0,0
0,30,0,0,0
0,45,0,0,0
上面的所有代码都是从 CSV 文件中一个一个地打印每个示例，虽然不错，但对于训练来说非常无用。
我在这里纠结的是你如何将这些单独的例子逐个加载到训练数据集中。例如，here's a notebook 我在 Udacity 深度学习课程中学习。我基本上想获取正在加载的 CSV 数据，然后将其放入 train_dataset 和 train_labels 之类的内容中:
def reformat(dataset, labels):
dataset = dataset.reshape((-1, image_size * image_size)).astype(np.float32)
# Map 2 to [0.0, 1.0, 0.0 ...], 3 to [0.0, 0.0, 1.0 ...]
labels = (np.arange(num_labels) == labels[:,None]).astype(np.float32)
return dataset, labels
train_dataset, train_labels = reformat(train_dataset, train_labels)
valid_dataset, valid_labels = reformat(valid_dataset, valid_labels)
test_dataset, test_labels = reformat(test_dataset, test_labels)
print('Training set', train_dataset.shape, train_labels.shape)
print('Validation set', valid_dataset.shape, valid_labels.shape)
print('Test set', test_dataset.shape, test_labels.shape)
我试过像这样使用 tf.train.shuffle_batch，但它莫名其妙地挂起：
for i in range(file_length):
# retrieve a single instance
example, label = sess.run([features, colRelevant])
example_batch, label_batch = tf.train.shuffle_batch([example, label], batch_size=file_length, capacity=file_length, min_after_dequeue=10000)
print(example, label)
总而言之，这是我的问题：
- 这个过程我错过了什么？
- 感觉我缺少一些关于如何正确构建输入管道的关键直觉。
- 有没有办法避免必须知道 CSV 文件的长度？
- 必须知道要处理的行数（上面的
for i in range(file_length)行代码）感觉很不优雅
- 必须知道要处理的行数（上面的
编辑： 一旦 Yaroslav 指出我可能在这里混淆了命令式和图形构造部分，它就开始变得更加清晰。我能够将以下代码整合在一起，我认为这更接近于从 CSV 训练模型时通常所做的（不包括任何模型训练代码）：
from __future__ import print_function
import numpy as np
import tensorflow as tf
import math as math
import argparse
parser = argparse.ArgumentParser()
parser.add_argument('dataset')
args = parser.parse_args()
def file_len(fname):
with open(fname) as f:
for i, l in enumerate(f):
pass
return i + 1
def read_from_csv(filename_queue):
reader = tf.TextLineReader(skip_header_lines=1)
_, csv_row = reader.read(filename_queue)
record_defaults = [[0],[0],[0],[0],[0]]
colHour,colQuarter,colAction,colUser,colLabel = tf.decode_csv(csv_row, record_defaults=record_defaults)
features = tf.stack([colHour,colQuarter,colAction,colUser])
label = tf.stack([colLabel])
return features, label
def input_pipeline(batch_size, num_epochs=None):
filename_queue = tf.train.string_input_producer([args.dataset], num_epochs=num_epochs, shuffle=True)
example, label = read_from_csv(filename_queue)
min_after_dequeue = 10000
capacity = min_after_dequeue + 3 * batch_size
example_batch, label_batch = tf.train.shuffle_batch(
[example, label], batch_size=batch_size, capacity=capacity,
min_after_dequeue=min_after_dequeue)
return example_batch, label_batch
file_length = file_len(args.dataset) - 1
examples, labels = input_pipeline(file_length, 1)
with tf.Session() as sess:
tf.initialize_all_variables().run()
# start populating filename queue
coord = tf.train.Coordinator()
threads = tf.train.start_queue_runners(coord=coord)
try:
while not coord.should_stop():
example_batch, label_batch = sess.run([examples, labels])
print(example_batch)
except tf.errors.OutOfRangeError:
print('Done training, epoch reached')
finally:
coord.request_stop()
coord.join(threads)
我认为您在这里混淆了命令式和图形构造部分。 tf.train.shuffle_batch操作创建了一个新的队列节点，单个节点可以用来处理整个数据集。所以我认为你挂了，因为你在 for 循环中创建了一堆 shuffle_batch 队列并且没有为它们启动队列 运行ners。
正常的输入管道用法如下所示：
- 将
shuffle_batch等节点添加到输入管道 - （可选，以防止无意中修改图形）完成图形
---图构造结束，命令式编程开始--
tf.start_queue_runnerswhile(True): session.run()
为了更具可扩展性（避免 Python GIL），您可以使用 TensorFlow 管道生成所有数据。但是，如果性能不是很重要，您可以使用 slice_input_producer. 将 numpy 数组连接到输入管道这是一个包含一些 Print 节点的示例，以查看发生了什么（Print 中的消息当节点为 运行)
tf.reset_default_graph()
num_examples = 5
num_features = 2
data = np.reshape(np.arange(num_examples*num_features), (num_examples, num_features))
print data
(data_node,) = tf.slice_input_producer([tf.constant(data)], num_epochs=1, shuffle=False)
data_node_debug = tf.Print(data_node, [data_node], "Dequeueing from data_node ")
data_batch = tf.batch([data_node_debug], batch_size=2)
data_batch_debug = tf.Print(data_batch, [data_batch], "Dequeueing from data_batch ")
sess = tf.InteractiveSession()
sess.run(tf.initialize_all_variables())
tf.get_default_graph().finalize()
tf.start_queue_runners()
try:
while True:
print sess.run(data_batch_debug)
except tf.errors.OutOfRangeError as e:
print "No more inputs."
你应该看到这样的东西
[[0 1]
[2 3]
[4 5]
[6 7]
[8 9]]
[[0 1]
[2 3]]
[[4 5]
[6 7]]
No more inputs.
“8、9”号没有填满整批，所以没有生产。 tf.Print 也被打印到 sys.stdout，因此它们会单独显示在终端中。
PS: 最小连接 batch 到手动初始化的队列是在 github issue 2193
此外，出于调试目的，您可能需要在会话中设置 timeout，这样您的 IPython 笔记本就不会在空队列出队时挂起。我在我的会话中使用这个辅助函数
def create_session():
config = tf.ConfigProto(log_device_placement=True)
config.gpu_options.per_process_gpu_memory_fraction=0.3 # don't hog all vRAM
config.operation_timeout_in_ms=60000 # terminate on long hangs
# create interactive session to register a default session
sess = tf.InteractiveSession("", config=config)
return sess
可扩展性说明：
tf.constant将您的数据副本内联到图表中。图定义的大小有 2GB 的基本限制，因此这是数据大小的上限- 您可以通过使用
v=tf.Variable并通过 运行 将v.assign_op与tf.placeholder放在右侧并将数据保存到其中来绕过该限制numpy 数组到占位符 (feed_dict) - 这仍然会创建两个数据副本，因此为了节省内存，您可以制作自己的
slice_input_producer版本，它在 numpy 数组上运行，并使用feed_dict[= 一次上传一行60=]
或者你可以试试这个，代码使用 pandas 和 numpy 将 Iris 数据集加载到 tensorflow 中，并在会话中打印一个简单的神经元输出。希望对大家有个基本的了解有帮助.... [我还没有添加一个热解码标签的方式]。
import tensorflow as tf
import numpy
import pandas as pd
df=pd.read_csv('/home/nagarjun/Desktop/Iris.csv',usecols = [0,1,2,3,4],skiprows = [0],header=None)
d = df.values
l = pd.read_csv('/home/nagarjun/Desktop/Iris.csv',usecols = [5] ,header=None)
labels = l.values
data = numpy.float32(d)
labels = numpy.array(l,'str')
#print data, labels
#tensorflow
x = tf.placeholder(tf.float32,shape=(150,5))
x = data
w = tf.random_normal([100,150],mean=0.0, stddev=1.0, dtype=tf.float32)
y = tf.nn.softmax(tf.matmul(w,x))
with tf.Session() as sess:
print sess.run(y)
你可以使用最新的 tf.data API :
dataset = tf.contrib.data.make_csv_dataset(filepath)
iterator = dataset.make_initializable_iterator()
columns = iterator.get_next()
with tf.Session() as sess:
sess.run([iteator.initializer])
如果有人来这里寻找一种简单的方法来读取 tf.estimator API 中绝对大的和分片的 CSV 文件，那么请看下面我的代码
CSV_COLUMNS = ['ID','text','class']
LABEL_COLUMN = 'class'
DEFAULTS = [['x'],['no'],[0]] #Default values
def read_dataset(filename, mode, batch_size = 512):
def _input_fn(v_test=False):
# def decode_csv(value_column):
# columns = tf.decode_csv(value_column, record_defaults = DEFAULTS)
# features = dict(zip(CSV_COLUMNS, columns))
# label = features.pop(LABEL_COLUMN)
# return add_engineered(features), label
# Create list of files that match pattern
file_list = tf.gfile.Glob(filename)
# Create dataset from file list
#dataset = tf.data.TextLineDataset(file_list).map(decode_csv)
dataset = tf.contrib.data.make_csv_dataset(file_list,
batch_size=batch_size,
column_names=CSV_COLUMNS,
column_defaults=DEFAULTS,
label_name=LABEL_COLUMN)
if mode == tf.estimator.ModeKeys.TRAIN:
num_epochs = None # indefinitely
dataset = dataset.shuffle(buffer_size = 10 * batch_size)
else:
num_epochs = 1 # end-of-input after this
batch_features, batch_labels = dataset.make_one_shot_iterator().get_next()
#Begins - Uncomment for testing only -----------------------------------------------------<
if v_test == True:
with tf.Session() as sess:
print(sess.run(batch_features))
#End - Uncomment for testing only -----------------------------------------------------<
return add_engineered(batch_features), batch_labels
return _input_fn
TF.estimator中的用法示例：
train_spec = tf.estimator.TrainSpec(input_fn = read_dataset(
filename = train_file,
mode = tf.estimator.ModeKeys.TRAIN,
batch_size = 128),
max_steps = num_train_steps)
2.0 兼容解决方案：此答案可能由上述线程中的其他人提供，但我将提供有助于社区的其他链接。
dataset = tf.data.experimental.make_csv_dataset(
file_path,
batch_size=5, # Artificially small to make examples easier to show.
label_name=LABEL_COLUMN,
na_value="?",
num_epochs=1,
ignore_errors=True,
**kwargs)
更多信息，请参考此Tensorflow Tutorial。