Luft: 10 सेकंड में 1 अरब डेटा पर क्वेरी करने वाले datastore के विकास की कहानी

मासिक औसत इवेंट की संख्या 10 अरब से अधिक होने वाले वातावरण में, कम समय में डेटा का विश्लेषण करके यूज़र behavior analysis (Cohort) करना ज़रूरी हो गया।

(उदाहरण: पिछले 6 महीनों में हमारे ऐप में हर महीने 1 लाख वॉन से अधिक खर्च करने वाली 30s की महिलाएँ → इनकी revisit rate)

यह ऐसी कहानी है जिसमें डेवलपर ने उस datastore को, जिसे अब तक सिर्फ इस्तेमाल किया था, खुद इम्प्लीमेंट किया।

यूज़र behavior analysis query को इम्प्लीमेंट करने के लिए…

• पहले से pre-calculate न किए गए metrics पर भी query चलनी चाहिए (+ नए प्रकार के analysis भी re-indexing के बिना संभव होने चाहिए)

• इवेंट डेटा को user-आधारित Group By करते समय High Cardinality Shuffle का bottleneck कम होना चाहिए

मौजूदा solution इस्तेमाल करें या अपना solution बनाएं, इस पर विचार किया गया

• Druid कहीं और इस्तेमाल हो रहा था, लेकिन Pre-Aggregation (सिर्फ गणना किए गए मान पढ़ने की विधि) की सीमा के कारण यह फीचर इम्प्लीमेंटेशन के लिए उपयुक्त नहीं था

• Snowflake या Redshift जैसे data warehouse बड़े पैमाने पर चलाए जा सकते हैं, लेकिन उनकी general-purpose प्रकृति के कारण लक्ष्य की तुलना में बहुत बड़ा cluster चलाना पड़ता, इसलिए वे महंगे थे

• Funnel, ID matching जैसी विविध ज़रूरतों को कवर करने के लिए SQL-आधारित DB की सीमाएँ थीं

आखिरकार datastore खुद बनाया गया

• Luft = शुरू से ही user ID-आधारित Group By किए गए यूज़र behavior analysis query को तेज़ी से चलाने के लिए optimized data store

• Golang के आधार पर बनाया गया

• कई दर्जन TB आकार के यूज़र डेटा का विश्लेषण 5 से कम nodes पर औसतन 3 सेकंड ~ अधिकतम 10 सेकंड के बीच

• सामान्य RDBMS से अलग, इसमें immutability है (ज़रूरत पड़ने पर उसी अवधि के डेटा को overwrite किया जाता है) → simple cluster design, complex page manager इम्प्लीमेंटेशन के बिना high performance, और मनचाहा data storage format design संभव

तकनीकी आधार को विस्तार से देखें

• TrailDB (storage engine) - user ID partitioning के लिए optimized time-series event storage Rowstore

→ मानों को dictionary-encode करके केवल उनके ID स्टोर किए जाते हैं

→ यूज़र इवेंट्स को समयक्रम में sort करके, पिछले इवेंट की तुलना में बढ़ा हुआ time value और बदले हुए columns ही स्टोर किए जाते हैं (क्योंकि अधिकांश user properties नहीं बदलतीं)

→ कोई index नहीं। हमेशा full scan करना पड़ता है।

→ लेकिन चौंकाने वाली compression ratio देता है (CSV 13GB → ~TrailDB 300mb)

→ time complexity O(n) है, इसलिए सोचा गया कि space complexity कम कर दी जाए

• LLVM (query engine)

→ लेकिन TrailDB सिर्फ OR-AND फ़ॉर्म के equals प्रदान करता है, और Go में parse की गई query को C, C++ में भेजना पड़ता है

→ पता चला कि PostgreSQL query को LLVM JiT में compile करता है

→ query में फीचर विस्तार अक्सर होता है, इसलिए C, C++ में लिखने से बढ़ने वाली development cost को रोका जा सकता है (Golang में सिर्फ LLVM IR बनाकर भेज दें, और C, C++ में JiT compile कराके चलाया जाए)

• computation layer भी खुद बनाई

→ MapReduce बहुत इस्तेमाल होता है, लेकिन Golang इस्तेमाल करने के कारण इसे नहीं अपना सके

→ Spark/Hadoop Long-running Job के लिए optimized हैं, इसलिए जोड़कर देखने पर भी performance अच्छी नहीं मिली

→ यह भी खुद बनाया गया → https://github.com/ab180/lrmr

→ gRPC + Protobuf + etcd का संयोजन, और Spark के परिचित design को काफी अपनाया गया

→ Resiliency छोड़ दें → अगर performance को चरम तक बढ़ा दिया जाए, तो failure होने पर भी शुरुआत से दोबारा करने में 10 सेकंड से कम लगेंगे

→ बड़े पैमाने के डेटा प्रोसेसिंग से buffer overflow अक्सर होता था (Backpressure), इसलिए इसे Pull-based Event Stream में बदला गया (Kafka, Armeria आदि में अपनाया गया)

• sharding भी खुद इम्प्लीमेंट की

→ shard = historical node

→ अगर partition की date range को sharding key value के रूप में इस्तेमाल करें?

→ हर query में time होता है → filtering आसान

→ एक ही time range में लगभग समान डेटा मात्रा होती है → data distribution आसान

→ distributed environment सुंदर नहीं होता…

→ node down हो जाए या नया जोड़ा जाए तो?

→ storage space भर जाए तो?

→ failure के कारण लोड एक ही node पर केंद्रित हो जाए तो?

→ Druid के Cost Function को customize करके partition date range जितनी पास हो और जितनी ज़्यादा overlap करे, Cost उतनी अधिक हो, ऐसा बनाया गया

→ shard availability के लिए नीचे के काम किए गए

→ shard information पर TTL लगाया और इसे periodic रूप से refresh किया (etcd)

 → S3 में partition स्टोर किए, और DynamoDB से partition list मैनेज की

मौजूदा production स्थिति

• सिर्फ 4 c5.2xlarge instances से 15 सेकंड के भीतर 500GB डेटा scan

आगे के लक्ष्य (या किए जाने वाले काम)

• real-time Funnel analysis को 10 से कम nodes वाले cluster पर करना चाहते हैं

• Spark support जोड़कर ML integration आदि को support करना चाहते हैं

• TrailDB को बदलने के लिए अपना column store (Ziegel) विकसित कर रहे हैं

→ SIMD और multi-core optimization

→ Bitmap Index से पहले ही user properties-आधारित filtering

[gera1d]

traildb मज़ेदार है। https://www.youtube.com/watch?v=-oPFxSwn0lM दिलचस्प है। वीडियो काफ़ी पुराना है, लेकिन शायद इस दौरान traildb में कोई बदलाव नहीं हुआ होगा।

[hiddenest]

अब देखा तो डेवलपर की ब्लॉग पोस्ट भी है,

https://engineering.ab180.co/stories/introducing-luft

TrailDB के बारे में पहली बार सुना, यह कुछ ऐसा है...

https://github.com/traildb/traildb