स्पार्क समर्थन पक्षियों के साथ विभाजन छिद्रण

Question

मैं एक बड़े डेटासेट के साथ काम कर रहा हूं, जिसे दो कॉलम - plant_name और tag_id द्वारा विभाजित किया गया है। दूसरे विभाजन - tag_id 200000 अद्वितीय मूल्य हैं और मैं ज्यादातर विशिष्ट tag_id मूल्यों से डेटा का उपयोग। यदि मैं निम्नलिखित स्पार्क कमांड का उपयोग करता हूं:

sqlContext.setConf("spark.sql.hive.metastorePartitionPruning", "true") 
sqlContext.setConf("spark.sql.parquet.filterPushdown", "true") 
val df = sqlContext.sql("select * from tag_data where plant_name='PLANT01' and tag_id='1000'") 

मुझे एक तेज प्रतिक्रिया की उम्मीद होगी क्योंकि यह एक विभाजन को हल करता है। हाइव और प्रेस्टो में इसमें सेकंड लगते हैं, हालांकि स्पार्क में यह घंटों तक चलता है।

वास्तविक डेटा एस 3 बाल्टी में आयोजित किया जाता है, और जब मैं एसक्यूएल क्वेरी सबमिट करता हूं, स्पार्क बंद हो जाता है और पहले हाइव मेटास्टोर (उनमें से 200000) से सभी विभाजन प्राप्त होते हैं, और फिर पूर्ण स्थिति को मजबूर करने के लिए refresh() पर कॉल करते हैं S3 वस्तु की दुकान में इन सभी फ़ाइलों की सूची (वास्तव में listLeafFilesInParallel बुला)।

यह दो ऑपरेशन इतने महंगे हैं, क्या ऐसी कोई सेटिंग है जो स्पार्क को पहले विभाजन को छीनने के लिए मिल सकती है - या तो मेटाडेटा स्टोर पर कॉल के दौरान या तुरंत बाद में?

Answer 1

बस एक विचार: HadoopFsRelation के लिए

स्पार्क API दस्तावेज़ कहते हैं, (https://spark.apache.org/docs/1.6.2/api/java/org/apache/spark/sql/sources/HadoopFsRelation.html)

"... जब हाइव शैली विभाजित फ़ाइल सिस्टम में संगृहीत टेबल से पढ़ने, यह पता करने में सक्षम है इनपुट निर्देशिकाओं के पथ से विभाजित जानकारी, और डेटा पढ़ने से पहले विभाजन छंटनी करें ... "

तो, मैं "listLeafFilesInParallel" लगता है कि एक समस्या नहीं हो सकता है।

ऐसा ही एक मुद्दा चिंगारी jira में पहले से ही है: https://issues.apache.org/jira/browse/SPARK-10673

"spark.sql.hive.verifyPartitionPath" झूठे और करने के लिए सेट के बावजूद, वहाँ प्रदर्शन में कोई प्रभाव नहीं है, मुझे लगता है मुद्दा हो सकता है अनियंत्रित विभाजन के कारण हुआ है। कृपया तालिका के विभाजन की सूची और सत्यापित करें कि सभी विभाजन पंजीकृत हैं। वरना, अपने विभाजन के रूप में इस लिंक में दिखाया गया है की वसूली:

Hive doesn't read partitioned parquet files generated by Spark

अद्यतन:

1. मैं उचित छत ब्लॉक आकार और पृष्ठ आकार लगता है, जबकि डेटा लेखन निर्धारित किया गया।

2. वर्णित विभाजनों के साथ एक ताजा हाइव तालिका बनाएं, और फ़ाइल प्रारूप को लकड़ी के रूप में विभाजित करें, इसे गतिशील विभाजन दृष्टिकोण का उपयोग करके गैर-विभाजित तालिका से लोड करें। (https://cwiki.apache.org/confluence/display/Hive/DynamicPartitions) एक सादे छत्ता क्वेरी चलाएं और फिर एक चिंगारी कार्यक्रम चलाकर की तुलना करें।

अस्वीकरण: मैं एक चिंगारी/छत विशेषज्ञ नहीं हूँ। समस्या दिलचस्प लग रही थी, और इसलिए जवाब दिया।

Answer 2

हाँ, चिंगारी विभाजन छंटाई का समर्थन करता है।

स्पार्क पार्टिशन निर्देशिकाओं (अनुक्रमिक या समांतर listLeafFilesInParallel) की सूची को पहली बार सभी विभाजनों के कैश बनाने के लिए करता है। उसी एप्लिकेशन में प्रश्न, जो डेटा स्कैन करते हैं, इस कैश का लाभ उठाते हैं। तो आप जो धीमापन देखते हैं वह इस कैश बिल्डिंग के कारण हो सकता है। बाद के प्रश्न जो डेटा स्कैन करते हैं, विभाजन को छीनने के लिए कैश का उपयोग करते हैं।

ये वे लॉग हैं जो कैश को पॉप्युलेट करने के लिए सूचीबद्ध विभाजन दिखाते हैं।

App > 16/11/14 10:45:24 main INFO ParquetRelation: Listing s3://test-bucket/test_parquet_pruning/month=2015-01 on driver 
App > 16/11/14 10:45:24 main INFO ParquetRelation: Listing s3://test-bucket/test_parquet_pruning/month=2015-02 on driver 
App > 16/11/14 10:45:24 main INFO ParquetRelation: Listing s3://test-bucket/test_parquet_pruning/month=2015-03 on driver 

ये लॉग दिखा रहे हैं कि छंटनी हो रही है।

App > 16/11/10 12:29:16 main INFO DataSourceStrategy: Selected 1 partitions out of 20, pruned 95.0% partitions. 

HiveMetastoreCatalog.scala में convertToParquetRelation और getHiveQlPartitions देखें।