Note: Before someone suggests existing solutions, allow me to tell you respectfully that I already looked at various solutions, my error message might be the same but not the problem and the current implementation itself so allow me to explain my issue
I am using multiprocessing to spawn multiple processes to scrape multiple URLs at once. In order to store data, I am using MySQL and using a single connection created in __main__ during the entire execution. After a certain period of time, my script gets stuck. When I run the query SELECT * FROM INFORMATION_SCHEMA.PROCESSLIST WHERE DB = "ecom_trends_db"; I find a SLEEP entry of various seconds. In order to deal with that I did the following when fetching links in the loop:
connection.ping(reconnect=True)
in the __main__ while I am fetching the connection the first time I did the following:
connection = get_connection(DB_HOST, DB_USER, DB_PASSWORD, DB_NAME)
with connection.cursor() as cursor:
cursor.execute('SET interactive_timeout = 180;')
cursor.execute('SET wait_timeout = 180;')
cursor.execute('SET net_read_timeout = 180;')
cursor.execute('SET GLOBAL connect_timeout = 180;')
connection.commit()
But still, things get stuck for a long time and I have to kill the script. If I do not kill the script, the SLEEP query time is increased and I just can't do any operation on the table where the data is being inserted. The entire code structure is given below. I am not sure whether this is something related to MySQL driver(pymysql) or something else.
import pymysql
def store_parsed(url, name, location, avg_review, sales, admirers, rating, review_count, year):
try:
if connection is not None:
with connection.cursor() as cursor:
logging.info('INSERT QUERY for URL = ' + url)
sql = 'INSERT INTO {} (url,name, location, average_review, sales, admirers, rating, review_count,etsy_since) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)'.format(
TABLE_PARSE)
cursor.execute(sql, (url, name, location, avg_review, sales, admirers, rating, review_count, year))
# Update the status of URL in shop_links
logging.info('UPDATE QUERY for URL = ' + url)
sql = "UPDATE {} set status = 3 where url = '{}' ".format(TABLE_FETCH, url)
cursor.execute(sql)
connection.commit()
print('Data Commit')
logging.info('Data Commit')
except pymysql.err.OperationalError:
print('Lost Connection in store_parsed. Pinging Now')
# connection.ping(True)
except Exception as ex:
print('Exception in store_parsed')
crash_date = time.strftime("%Y-%m-%d %H:%m:%S")
crash_string = "".join(traceback.format_exception(etype=type(ex), value=ex, tb=ex.__traceback__))
exception_string = '[' + crash_date + '] - ' + crash_string + '\n'
print(exception_string)
logging.warning('Exception in store_parsed')
logging.error(exception_string)
def parse(url):
# parsing stuff
store_parsed(url, name, location, avg_review, sales, admirers, rating, review_count, year)
def get_links(size=100):
total_links = []
_links = []
try:
if connection is not None:
connection.ping(reconnect=True)
with connection.cursor() as cursor:
sql = 'SELECT DISTINCT(url) from {} WHERE STATUS = 0 LIMIT {}'.format(TABLE_FETCH, size)
cursor.execute(sql)
links = cursor.fetchall()
for link in links:
# total_links.append('https://hotpads.com{}'.format(link['url'].strip()))
total_links.append(link['url'].strip())
_links.append(link['url'].strip())
print('Total = {}'.format(len(_links)))
format_strings = ','.join(['%s'] * len(links))
if len(total_links) > 0:
if connection is not None:
with connection.cursor() as cursor:
sql = " UPDATE " + TABLE_FETCH + " set status = 1 WHERE url IN (%s)" % format_strings
cursor.execute(sql, tuple(_links))
connection.commit()
print('Affected UPDATED ROWS Rows:- {0}'.format(cursor.rowcount))
def get_connection(host, user, password, db_name):
connection = None
try:
connection = pymysql.connect(host=host,
user=user,
password=password,
db=db_name,
charset='utf8',
max_allowed_packet=1073741824,
cursorclass=pymysql.cursors.DictCursor)
print('Connected')
except Exception as ex:
print(str(ex))
finally:
return connection
if __name__ == '__main__':
connection = None
connection = get_connection(DB_HOST, DB_USER, DB_PASSWORD, DB_NAME)
with connection.cursor() as cursor:
cursor.execute('SET interactive_timeout = 180;')
cursor.execute('SET wait_timeout = 180;')
cursor.execute('SET net_read_timeout = 180;')
cursor.execute('SET GLOBAL connect_timeout = 180;')
connection.commit()
while True:
print('Getting Links')
logging.info('Getting Links')
links = get_links(LIMIT)
if len(links) == 0:
break
if len(links) > 0:
with Pool(POOL_COUNT) as p:
result = p.map(parse, links)
Hope I clarified my issue. Please help!!! Should I create multiple connections but it will be another issue. How do I deal with it?
After decreasing timeout it was throwing the following exception:
Data Stored
=============END====================
Getting Links
[2022-08-16 20:08:17] - Traceback (most recent call last):
File "/Users/AdnanAhmad/Data/anaconda3/lib/python3.7/site-packages/pymysql/connections.py", line 732, in _read_bytes
data = self._rfile.read(num_bytes)
File "/Users/AdnanAhmad/Data/anaconda3/lib/python3.7/socket.py", line 589, in readinto
return self._sock.recv_into(b)
ConnectionResetError: [Errno 54] Connection reset by peer
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "parse_db.py", line 69, in get_links
cursor.execute(sql)
File "/Users/AdnanAhmad/Data/anaconda3/lib/python3.7/site-packages/pymysql/cursors.py", line 148, in execute
result = self._query(query)
File "/Users/AdnanAhmad/Data/anaconda3/lib/python3.7/site-packages/pymysql/cursors.py", line 310, in _query
conn.query(q)
File "/Users/AdnanAhmad/Data/anaconda3/lib/python3.7/site-packages/pymysql/connections.py", line 548, in query
self._affected_rows = self._read_query_result(unbuffered=unbuffered)
File "/Users/AdnanAhmad/Data/anaconda3/lib/python3.7/site-packages/pymysql/connections.py", line 775, in _read_query_result
result.read()
File "/Users/AdnanAhmad/Data/anaconda3/lib/python3.7/site-packages/pymysql/connections.py", line 1156, in read
first_packet = self.connection._read_packet()
File "/Users/AdnanAhmad/Data/anaconda3/lib/python3.7/site-packages/pymysql/connections.py", line 692, in _read_packet
packet_header = self._read_bytes(4)
File "/Users/AdnanAhmad/Data/anaconda3/lib/python3.7/site-packages/pymysql/connections.py", line 740, in _read_bytes
"Lost connection to MySQL server during query (%s)" % (e,),
pymysql.err.OperationalError: (2013, 'Lost connection to MySQL server during query ([Errno 54] Connection reset by peer)')
Traceback (most recent call last):
File "parse_db.py", line 250, in <module>
reset_count(connection)
File "parse_db.py", line 17, in reset_count
cursor.execute(sql)
File "/Users/AdnanAhmad/Data/anaconda3/lib/python3.7/site-packages/pymysql/cursors.py", line 148, in execute
result = self._query(query)
File "/Users/AdnanAhmad/Data/anaconda3/lib/python3.7/site-packages/pymysql/cursors.py", line 310, in _query
conn.query(q)
File "/Users/AdnanAhmad/Data/anaconda3/lib/python3.7/site-packages/pymysql/connections.py", line 547, in query
self._execute_command(COMMAND.COM_QUERY, sql)
File "/Users/AdnanAhmad/Data/anaconda3/lib/python3.7/site-packages/pymysql/connections.py", line 793, in _execute_command
raise err.InterfaceError(0, "")
pymysql.err.InterfaceError: (0, '')
No comments:
Post a Comment