Comment lire *réellement* des données CSV dans TensorFlow ?

Question

Je suis relativement nouveau dans le monde de TensorFlow, et je suis assez perplexe quant à la façon dont on peut en fait lire des données CSV en un exemple utilisable/étiqueter des tenseurs dans TensorFlow. L'exemple de la Tutoriel TensorFlow sur la lecture de données CSV est assez fragmentée et ne vous permet que partiellement de vous entraîner sur des données CSV.

Voici le code que j'ai assemblé, basé sur ce tutoriel 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")

Et voici un bref exemple du fichier CSV que je charge - des données assez basiques - 4 colonnes de caractéristiques, et 1 colonne d'étiquettes :

0,0,0,0,0
0,15,0,0,0
0,30,0,0,0
0,45,0,0,0

Tout ce que le code ci-dessus fait est imprimer chaque exemple du fichier CSV, un par un qui, bien qu'agréable, est plutôt inutile pour la formation.

Ce que je ne comprends pas, c'est comment transformer ces exemples individuels, chargés un par un, en un ensemble de données de formation. Par exemple, voici un carnet de notes sur lequel je travaillais dans le cours Udacity Deep Learning. En gros, je veux prendre les données CSV que je charge, et les placer dans quelque chose comme train_dataset et 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)

J'ai essayé d'utiliser tf.train.shuffle_batch comme ça, mais ça se bloque inexplicablement :

  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)

Donc, pour résumer, voici mes questions :

• Qu'est-ce qui me manque dans ce processus ?
  • J'ai l'impression qu'il y a une intuition clé qui me manque sur la façon de construire correctement un pipeline d'entrée.
• Existe-t-il un moyen d'éviter de devoir connaître la longueur du fichier CSV ?
  • Il semble assez inélégant de devoir connaître le nombre de lignes que l'on veut traiter (la fonction for i in range(file_length) ligne de code ci-dessus)

---

Edit : Dès que Yaroslav m'a fait remarquer que je mélangeais probablement les parties impératives et de construction de graphe, les choses ont commencé à devenir plus claires. J'ai pu rassembler le code suivant, qui, je pense, est plus proche de ce qui est généralement fait lors de l'entraînement d'un modèle à partir de CSV (à l'exclusion de tout code d'entraînement de modèle) :

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)

Answer 1

Je pense que vous mélangez les parties impératives et de construction de graphe ici. L'opération tf.train.shuffle_batch crée un nouveau nœud de file d'attente, et un seul nœud peut être utilisé pour traiter l'ensemble des données. Donc, je pense que vous êtes suspendu parce que vous avez créé un tas de nœuds de queue. shuffle_batch des files d'attente dans votre boucle for et n'a pas démarré de file d'attente pour eux.

L'utilisation normale du pipeline d'entrée ressemble à ceci :

1. Ajoutez des nœuds comme shuffle_batch vers le pipeline d'entrée
2. (facultatif, pour éviter toute modification involontaire du graphique) finaliser le graphique

--- fin de la construction des graphes, début de la programmation impérative --

3. tf.start_queue_runners
4. while(True): session.run()

Pour être plus évolutif (et éviter le GIL de Python), vous pourriez générer toutes vos données à l'aide du pipeline TensorFlow. Cependant, si les performances ne sont pas critiques, vous pouvez connecter un tableau numpy à un pipeline d'entrée en utilisant la commande slice_input_producer. Voici un exemple avec quelques Print pour voir ce qui se passe (messages en Print aller à stdout quand le noeud est exécuté)

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."

Vous devriez voir quelque chose comme ceci

[[0 1]
 [2 3]
 [4 5]
 [6 7]
 [8 9]]
[[0 1]
 [2 3]]
[[4 5]
 [6 7]]
No more inputs.

Les numéros "8, 9" n'ont pas rempli le lot complet, ils n'ont donc pas été produits. Aussi tf.Print sont imprimés dans sys.stdout, donc ils apparaissent séparément dans le Terminal pour moi.

PS : un minimum de connexion batch à une file d'attente initialisée manuellement est en numéro 2193 de github

De plus, pour des raisons de débogage, vous pouvez définir le paramètre timeout sur votre session afin que votre notebook IPython ne soit pas bloqué par des files d'attente vides. J'utilise cette fonction d'aide pour mes sessions

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

Notes sur l'évolutivité :

1. tf.constant copie en ligne de vos données dans le graphique. Il y a une limite fondamentale de 2GB sur la taille de la définition du graphique, donc c'est une limite supérieure sur la taille des données.
2. Vous pouvez contourner cette limite en utilisant v=tf.Variable et d'y sauvegarder les données en exécutant v.assign_op avec un tf.placeholder sur le côté droit et en alimentant le tableau numpy à l'emplacement ( feed_dict )
3. Cela crée toujours deux copies de données, donc pour économiser de la mémoire vous pouvez créer votre propre version de slice_input_producer qui opère sur des tableaux numpy, et télécharge les rangées une par une en utilisant feed_dict

Answer 2

Vous pouvez aussi essayer ceci : le code charge l'ensemble de données Iris dans tensorflow en utilisant pandas et numpy et une sortie simple à un neurone est imprimée dans la session. J'espère que cela vous aidera à avoir une compréhension de base. .... [ Je n'ai pas ajouté la manière de décoder les étiquettes à chaud].

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)

Answer 3

Si quelqu'un est venu ici à la recherche d'un moyen simple pour lire des fichiers CSV très volumineux et shardés dans l'API de tf.estimator, veuillez voir mon code ci-dessous.

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

Exemple d'utilisation dans 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)

Answer 4

Vous pouvez utiliser la dernière API tf.data :

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])

Answer 5

2.0 Solution compatible : Cette réponse pourrait être fournie par d'autres dans le fil de discussion ci-dessus, mais je vais fournir des liens supplémentaires qui aideront la communauté.

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)

Pour plus d'informations, veuillez consulter le site suivant Tutoriel Tensorflow .