python - 下载文件并推送到 MySQL，在 Python 中不会超时

Question

我下载了一个大的 CSV 远程文件，并希望将所有行推送到 MySQL 中。我使用 csv.reader 来解析远程文件。我将这些行以 1000 为批处理添加到 MySQL 中。

问题是与对等方的连接在 5 分钟后超时，虽然文件可以在不到一分钟内下载完毕，但推送到 MySQL 需要的时间更长。

有没有办法让下载作业和推送作业异步工作，以便与对等点的连接不等待 mySQL 约束？

我想避免

1. 如果不需要，将完整文件下载到内存中
2. 第一行下载完成后就开始推送到 mysql
3. 不得不处理临时文件

基本上，我希望我的 python 脚本执行类似 curl file | 的操作。 my_script_that_pushes_values.sh。

这是我所做的说明:

csvReader = csv.reader(distantfile)
valuesBuffer = []
for row in csvReader:
  valuesBuffer.append(getValues(row))
  if len(valuesBuffer) % 1000 = 0:
    pushValuesIntoMySQL(valuesBuffer)
    valuesBuffer = []
pushValuesIntoMySQL(valuesBuffer)

Answer 1

我会将整个文件复制到您的服务器，然后使用 LOAD DATA LOCAL INFILE因为它支持 csv 输入:

LOAD DATA INFILE 'data.txt' INTO TABLE tbl_name
  FIELDS TERMINATED BY ',' ENCLOSED BY '"'
  LINES TERMINATED BY '\r\n'
  IGNORE 1 LINES;

<小时/>

如果您不喜欢这个解决方案，您可以使用 mysql_ping() (希望您使用的连接器支持它)自动重新连接。

Checks whether the connection to the server is working. If the connection has gone down and auto-reconnect is enabled an attempt to reconnect is made. If the connection is down and auto-reconnect is disabled, mysql_ping() returns an error.

<小时/>

如果您遇到可以下载文件但由于 MySQL 延迟而超时的问题，您可以在 two threads 中运行它。并通过 queue 同步它:

# Prepare queue and end signaling handler
q = queue.Queue()
done = threading.Event()

# Function that fetches items from q and puts them into db after
# certain amount is reached
def store_db():
    items=[]

    # Until we set done
    while not done.is_set():
        try:
            # We may have 500 records and thread be done... prevent deadlock
            items.append(q.get(timeout=5))
            if len(items) > 1000:
                insert_into(items)
                items = []
            q.task_done()
         # If you wait longer then 5 seconds < exception
         except queue.Empty: pass

    if items:
        insert_into(items)

# Fetch all data in a loop
def continous_reading():
    # Fetch row
    q.put(row)

# Start storer thread
t = threading.Thread(target=store_db)
t.daemon = True
t.start()

continous_reading()
q.join() # Wait for all task to be processed
done.set() # Signal store_db that it can terminate
t.join() # to make sure the items buffer is stored into the db