In python, I'm facing issue with multithreading the code to retrieve.
I'm able to retrieve the data parallelly, however there is memory leak happening in the process of retrieving and it stops the process.
I wanted to achieve multithreading to do parallel collect the data without memory leak.
Please let me know, if any logic to change.
code:
import boto3, json, os
import base64
from datetime import datetime, timezone
import sys
import time
import threading
Shard_Iterator_ids=[]
def initial_shard_iterator(shard_id):
#Get Shard_Iterator using Shards_list
Shard_Iterator=kinesis.get_shard_iterator(
StreamName=Stream_name,
ShardId=shard_id,
ShardIteratorType=Type
)
Shard_Iterator_Id=Shard_Iterator['ShardIterator']
Shard_Iterator_Ids.append(Shard_Iterator_Id)
def get_record(Shard_Iterator_Id):
while True:
print(Shard_Iterator_Id)
Raw_data=kinesis.get_records(
ShardIterator=Shard_Iterator_Id,
Limit=100
)
if "NextShardIterator" in Raw_data:
#It get Shard_Iterator_Id to retrieve data again.
Shard_Iterator_Id=Raw_data['NextShardIterator']
try:
Json_data=Raw_data['Records']
print(Json_data)
except Exception as e:
pass
else:
pass
while True:
initial_shard_iterator(shard_id)
for i in range(0, len(Shard_Iterator_Ids)):
Shard_Iterator_id=Shard_Iterator_Ids[i]
t1=threading.Thread(target=get_record, args=(Shard_Iterator_id,))
t1.start()
Thankyou