解码的CSV文件出错:InvalidArgumentError:未引用的字段内部不能有引号/CRLF

Question

我使用的是Python 3.7.7，tensorflow io版本：0.17.0，tensorflow版本：2.4.0。我想从Kafka中读取CSV文件，并使用这些数据来训练模型。整个代码可以在上找到https://www.tensorflow.org/io/tutorials/kafka和https://github.com/tensorflow/io/blob/master/docs/tutorials/kafka.ipynb.

...
def decode_kafka_item(item):
message = tf.io.decode_csv(item.message, [[0.0] for i in range(NUM_COLUMNS)])
key = tf.strings.to_number(item.key)
return (message, key)
BATCH_SIZE=64
SHUFFLE_BUFFER_SIZE=64
train_ds = tfio.IODataset.from_kafka('susy-train', partition=0, offset=0)
train_ds = train_ds.shuffle(buffer_size=SHUFFLE_BUFFER_SIZE)
train_ds = train_ds.map(decode_kafka_item)
train_ds = train_ds.batch(BATCH_SIZE)
...
model.fit(train_ds, epochs=EPOCHS)

发出最后一行时的错误为：

InvalidArgumentError：未引用的字段内部不能有引号/CRLF[[｛node DecodeCSV｝｝]][[IteratorGetNext]][Op:__推理_训练_功能_956]

函数调用堆栈：列车功能

出现类似错误的问题有不同的代码，没有得到回答。

提前谢谢！

Answer 1

解决方案：
在从生产者端发送消息时，尝试替换所有CR('\r'(和LF('\n'(

因此Producer端的代码应该是这样的：

def write_to_kafka(topic_name, items):
count=0
producer = KafkaProducer(bootstrap_servers=['127.0.0.1:9092'])
for message, key in items:
message=message.split('r')[0] #Removing all the 'r' from the string
producer.send(topic_name, key=key.encode('utf-8'), value=message.encode('utf-8')).add_errback(error_callback)
count+=1
producer.flush()
print("Wrote {0} messages into topic: {1}".format(count, topic_name))

我的猜测：当我打印出制片人发送的消息时，我才知道这件事，消息看起来像这样：b'0,0,1,0,0,0r'由于错误显示：

未加引号的字段内部不能有引号/CRLF所以我只是更换了所有的CRLF

我在Windows机器上遇到了同样的问题，我在网上看到了一些答案，在Linux上也有同样的例子，我猜这只是Windows在csv文件中添加CRLF的问题，但这只是我的猜测，我不确定确切的原因。