Блог

Я пытаюсь найти более быстрый способ вставки данных в мою таблицу, в таблице должно получиться более 100 миллионов строк, я запускаю свой код почти 24 часа, и в настоящее время в таблицу введено только 9 миллионов строк, и она все еще выполняется.

В настоящее время мой код одновременно считывает 300 CSV-файлов и сохраняет данные в списке, он фильтруется на наличие повторяющихся строк, затем я использую цикл for для размещения записи в списке в виде кортежа и обновления таблицы по одному кортежу за раз. Этот метод просто занимает слишком много времени, есть ли способ для меня массово вставлять все строки? Я пытался искать в Интернете, но методы, которые я читаю, похоже, не помогают в моей ситуации.

Большое спасибо,

Дэвид

 import glob
import os
import csv
import mysql.connector

# MYSQL logon
mydb = mysql.connector.connect(
    host="localhost",
    user="David",
    passwd="Sword",
    database="twitch"
)
mycursor = mydb.cursor()

# list for strean data file names
streamData=[]

# This function obtains file name list from a folder, this is to open files 
in other functions
def getFileNames():
    global streamData
    global topGames

    # the folders to be scanned
    #os.chdir("D://UG_Project_Data")
    os.chdir("E://UG_Project_Data")
    # obtains stream data file names
    for file in glob.glob("*streamD*"):
        streamData.append(file)
    return

# List to store stream data from csv files
sData = []
# Function to read all streamData csv files and store data in a list
def streamsToList():
    global streamData
    global sData

    # Same as gamesToList
    index = len(streamData)
    num = 0
    theFile = streamData[0]
    for x in range(index):
        if (num == 301):
            filterStreams(sData)
            num = 0
            sData.clear()
        try:
            theFile = streamData[x]
            timestamp = theFile[0:15]
            dateTime = timestamp[4:8] "-" timestamp[2:4] "-" timestamp[0:2] "T" timestamp[9:11] ":" timestamp[11:13] ":" timestamp[13:15] "Z"
            with open (theFile, encoding="utf-8-sig") as f:
                reader = csv.reader(f)
                next(reader) # skip header
                for row in reader:
                    if (row != []):
                        col1 = row[0]
                        col2 = row[1]
                        col3 = row[2]
                        col4 = row[3]
                        col5 = row[4]
                        col6 = row[5]
                        col7 = row[6]
                        col8 = row[7]
                        col9 = row[8]
                        col10 = row[9]
                        col11 = row[10]
                        col12 = row[11]
                        col13 = dateTime
                        temp = col1, col2, col3, col4, col5, col6, col7, col8, col9, col10, col11, col12, col13
                        sData.append(temp)
        except:
            print("Problem file:")
            print(theFile)
        print(num)
        num  =1
    return

def filterStreams(self):
    sData = self
    dataSet = set(tuple(x) for x in sData)
    sData = [ list (x) for x in dataSet ]
    return createStreamDB(sData)

# Function to create a table of stream data
def createStreamDB(self):
    global mydb
    global mycursor
    sData = self
    tupleList = ()
    for x in sData:
        tupleList = tuple(x)
        sql = "INSERT INTO streams (id, user_id, user_name, game_id, community_ids, type, title, viewer_count, started_at, language, thumbnail_url, tag_ids, time_stamp) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
        val = tupleList
        try:
            mycursor.execute(sql, val)
            mydb.commit()
        except:
            test = 1
    return

if __name__== '__main__':
    getFileNames()
    streamsToList()
    filterStreams(sData)
  

Если некоторые из ваших строк завершаются успешно, а некоторые завершаются неудачей, вы хотите, чтобы ваша база данных осталась в поврежденном состоянии? если нет, попробуйте выполнить фиксацию вне цикла. вот так:

 for x in sData:
    tupleList = tuple(x)
    sql = "INSERT INTO streams (id, user_id, user_name, game_id, community_ids, type, title, viewer_count, started_at, language, thumbnail_url, tag_ids, time_stamp) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
    val = tupleList
    try:
        mycursor.execute(sql, val)
    except:
        # do some thing
        pass
try:
    mydb.commit()
except:
    test = 1
  

И если вы этого не сделаете. попробуйте загрузить ваш cvs файл непосредственно в ваш mysql.

 LOAD DATA INFILE "/home/your_data.csv"
INTO TABLE CSVImport
COLUMNS TERMINATED BY ','
OPTIONALLY ENCLOSED BY '"'
LINES TERMINATED BY 'n'
IGNORE 1 LINES;
  

Также, чтобы вам было понятнее. Я определил три способа вставки этих данных, если вы настаиваете на использовании python, поскольку у вас есть некоторая обработка ваших данных.

Плохой способ

 In [18]: def inside_loop(): 
    ...:     start = time.time() 
    ...:     for i in range(10000): 
    ...:         mycursor = mydb.cursor() 
    ...:         sql = "insert into t1(name, age)values(%s, %s)" 
    ...:         try: 
    ...:             mycursor.execute(sql, ("frank", 27)) 
    ...:             mydb.commit() 
    ...:         except: 
    ...:             print("Failure..") 
    ...:     print("cost :{}".format(time.time() - start)) 
    ...: 
  

Затраты времени:

 In [19]: inside_loop()                                                                                                                                                                                                                        
cost :5.92155909538269 
  

Хороший способ

 In [9]: def outside_loop(): 
   ...:     start = time.time() 
   ...:     for i in range(10000): 
   ...:         mycursor = mydb.cursor() 
   ...:         sql = "insert into t1(name, age)values(%s, %s)" 
   ...:         try: 
   ...:             mycursor.execute(sql, ["frank", 27]) 
   ...:         except: 
   ...:             print("do something ..") 
   ...:              
   ...:     try: 
   ...:         mydb.commit() 
   ...:     except: 
   ...:         print("Failure..") 
   ...:     print("cost :{}".format(time.time() - start))
  

Затраты времени:

 In [10]: outside_loop()                                                                                                                                                                                                                       
cost :0.9959311485290527
  

Возможно, все еще существует какой-то лучший способ, даже лучший. (т. е. используйте pandas для обработки ваших данных. и попробуйте перепроектировать свою таблицу …)

Возможно, вам понравится моя презентация Быстро загружайте данные! в которой я сравнил различные методы вставки объемных данных и провел тесты, чтобы определить, какой метод был самым быстрым.

Вставлять по одной строке за раз, совершая транзакцию для каждой строки, — это наихудший способ, которым вы можете это сделать.

Использование файла ЗАГРУЗКИ ДАННЫХ является самым быстрым с большим отрывом. Хотя есть некоторые изменения конфигурации, которые необходимо внести в экземпляр MySQL по умолчанию, чтобы он мог работать. Прочитайте документацию MySQL о параметрах secure_file_priv и local_infile .

Даже без использования LOAD DATA INFILE вы можете сделать намного лучше. Вы можете вставлять несколько строк для каждой вставки и выполнять несколько инструкций INSERT для каждой транзакции.

Однако я бы не стал пытаться вставить все 100 миллионов строк за одну транзакцию. Моя привычка — фиксировать примерно раз в 10 000 строк.