تقنية التدفق في كتابة ملفات JSON بلغة Python

عندما يتطلب مشروعك إنشاء ملف JSON ضخم، يمكن أن تواجه تحديات فيما يتعلق بالذاكرة ومساحة التخزين عند استخدام الطريقة التقليدية لتفريغ البيانات باستخدام json.dump(). تكون القائمة الكبيرة جداً بحيث لا يمكن للذاكرة ومساحة التبديل استيعابها قبل تفريغها. هل هناك وسيلة لتدفقها إلى ملف JSON؟

نعم، يمكنك تحقيق ذلك باستخدام تقنيات التدفق أو Streaming. بدلاً من تخزين القائمة الكاملة في الذاكرة قبل تفريغها، يمكنك كتابة البيانات إلى الملف تدفقياً وبتشغيل ملف JSON بالتدفق. تسمح لك مكتبة json في Python بفعل ذلك بسهولة.

للقيام بذلك، يمكنك استخدام json.JSONEncoder لتحويل البيانات إلى سلسلة JSON ومن ثم كتابتها إلى الملف بشكل تدفقي. على سبيل المثال، يمكنك استخدام مثل هذا الكود:

pythonCopy code
import json

class JSONStreamWriter:
    def __init__(self, filename):
        self.filename = filename

    def write_data(self, data):
        with open(self.filename, 'a', encoding='utf-8') as file:
            json.dump(data, file, ensure_ascii=False)
            file.write('\n')  # إضافة سطر جديد بعد كل تسجيل

# استخدام الكلاس
stream_writer = JSONStreamWriter('output.json')

# تحضير البيانات
large_data = [...]  # قائمة البيانات الكبيرة

# كتابة البيانات تدفقيا
for chunk in large_data:
    stream_writer.write_data(chunk)

تحتفظ هذه الطريقة بتأثير استهلاك الذاكرة منخفضًا نسبياً، حيث يتم كتابة البيانات إلى الملف بشكل فوري دون الحاجة إلى تخزينها ككل. كما يتيح لك إدراج تسجيلات في الملف تدفقياً دون التأثير الكبير على الموارد.

بالتأكيد، يمكننا توسيع نقاشنا لفهم المزيد حول كيفية تنفيذ هذه العملية بشكل أفضل وفهم التحديات المحتملة وكيفية التعامل معها.

في البداية، يجب أن نفهم أن الهدف من التدفق أو التدفق العكسي هو تجنب حدوث تأثير كبير على الذاكرة عند التعامل مع بيانات ضخمة. تقوم العديد من مكتبات Python بدعم تدفق البيانات، ولكن يمكن استخدام مكتبة ijson للتحكم بشكل أكبر في تحليل الملفات الكبيرة.

لتنفيذ ذلك، يمكنك استخدام ijson بمزجها مع مولدات Python لقراءة البيانات تدفقيًا وكتابتها إلى الملف. اليك مثال:

pythonCopy code
import ijson

class JSONStreamWriter:
    def __init__(self, filename):
        self.filename = filename

    def write_data(self, data):
        with open(self.filename, 'a', encoding='utf-8') as file:
            json.dump(data, file, ensure_ascii=False)
            file.write('\n')

def read_large_json(json_file):
    with open(json_file, 'r', encoding='utf-8') as file:
        parser = ijson.parse(file)
        for prefix, event, value in parser:
            yield {prefix: value}

# مثال على الاستخدام
input_json = 'input.json'
output_json = 'output.json'

stream_writer = JSONStreamWriter(output_json)

for chunk in read_large_json(input_json):
    stream_writer.write_data(chunk)

في هذا المثال، تستخدم ijson لتحليل الملف بشكل تدفقي ومن ثم يتم كتابة كل تشكيلة JSON إلى الملف النهائي. هذا يتيح لك التعامل مع ملفات JSON ذات حجم كبير دون الحاجة إلى تحميل البيانات بأكملها في الذاكرة.

يجب أن تأخذ في اعتبارك أن الأداء قد يختلف حسب نوع البيانات وحجمها، ويمكنك تحسين الكود بمراعاة متطلبات مشروعك الخاص.