Elasticsearch + wydajność Apache Spark

Question

Próbuję użyć iskry Apache do kwerendy moich danych w Elasticsearch, ale moje zadanie iskry trwa około 20 godzin, aby wykonać agregację i nadal działa. To samo zapytanie w ES trwa około 6 sekund.

Rozumiem, że dane muszą zostać przeniesione z klastra Elasticsearch do mojego klastra i niektórych danych w Spark.

Dane w moim indeksie ES wynoszą ok. 300 milionów dokumentów i każdy dokument ma około 400 pól (1,4 TB).

Posiadam 3-węzłowy klaster iskier (1 master, 2 workers) z 60 GB pamięci i 8 rdzeniami w sumie.

Czas potrzebny na uruchomienie jest nie do zaakceptowania, czy istnieje sposób na przyspieszenie pracy z iskrownikiem?

Oto moja konfiguracja iskra:

SparkConf sparkConf = new SparkConf(true).setAppName("SparkQueryApp") 
       .setMaster("spark://10.0.0.203:7077")  
       .set("es.nodes", "10.0.0.207") 
       .set("es.cluster", "wp-es-reporting-prod")    
       .setJars(JavaSparkContext.jarOfClass(Demo.class)) 
       .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") 
       .set("spark.default.parallelism", String.valueOf(cpus * 2)) 
       .set("spark.executor.memory", "8g"); 

Edited

SparkContext sparkCtx = new SparkContext(sparkConf); 

    SQLContext sqlContext = new SQLContext(sparkCtx); 
    DataFrame df = JavaEsSparkSQL.esDF(sqlContext, "customer-rpts01-201510/sample"); 

    DataFrame dfCleaned = cleanSchema(sqlContext, df); 

    dfCleaned.registerTempTable("RPT"); 

    DataFrame sqlDFTest = sqlContext.sql("SELECT agent, count(request_type) FROM RPT group by agent"); 

    for (Row row : sqlDFTest.collect()) { 
     System.out.println(">> " + row); 
    } 

Answer 1

Wyjaśniałem, co się dzieje, w zasadzie próbowałem manipulować schematem szkieletu danych, ponieważ mam pewne pola z kropką np. User.firstname. To wydaje się powodować problem w fazie zbierania iskry. Aby rozwiązać ten problem, musiałem po prostu ponownie zindeksować moje dane, aby moje pola nie zawierały kropki, ale podkreślenia, np. Nazwa_użytkownika.

Answer 2

Obawiam się, że nie można wykonać grupę o ponad 1,4 TB tylko 120 GB pamięci RAM i całkowitej osiągnąć dobrą wydajność. DF spróbuje załadować wszystkie dane w pamięci/na dysku i dopiero wtedy wykona grupę według. Nie sądzę, że w tej chwili złącze spark/ES tłumaczy składnię sql w języku zapytań ES.