Блог

Я получаю некоторые входные данные из восходящего потока, которые представляют собой sql-запросы в строковой форме, пути к входным файлам и имя представления таблицы. Для улучшения времени выполнения в кластере я хочу кэшировать свой набор данных для одного запроса и повторно использовать тот же набор данных для других запросов.

Например, рассмотрим следующий IPL набора данных.

 # player_name #  |  # team #  |  # country #  |  # runs #

sachin           |   csk      |    IND        |   12000
sehwag           |   DC       |    IND        |   18000
MOEN             |   RCB      |    UK         |   10000
WATSON           |   CSK      |    AUS        |   11000
  

и так далее…..

У меня есть следующие запросы.

1. Выберите команду, суммируйте (выполняйте) как total_run из IPL, где выполняется> 12000 групп по команде.
2. Выберите country, sum (runs) как total_run из IPL, где выполняется> 12000 групп по странам.

оба этих запроса используют один и тот же отфильтрованный набор данных, который выполняется> 12000, и может быть n таких запросов, которые могут использовать тот же результат, что и выше. как я могу кэшировать набор данных после фильтрации здесь?таким образом, одно и то же вычисление выполняется не более одного раза.

Примечание: мы можем легко сделать это, если мы используем API фрейма данных, разделив запрос.

пример —

 Dataset<Row> result1 = df.where("runs > 12000").select("team", "run")
                        .groupby("team").agg(count("run").as(""total_run));

Dataset<Row> result2 = df.where("runs > 12000").select("country", "run") 
                        .groupby("country").agg(count("run").as(""total_run));
  

мы можем разбить его следующим образом….

     Dataset<Row> cachedResult = df.where("runs > 12000").select("team", "run");
    
    Dataset<Row> result1 = cachedResult.groupby("team").agg(count("run").as(""total_run));
    
    Dataset<Row> result2 = df.where("runs > 12000").select("country", "run")
                            .groupby("country").agg(count("run").as(""total_run));
  

просто мои вопросы,

1. как мы можем добиться кэширования и других подобных операций, когда у нас есть запрос spark sql в строковой форме?

2. нужно ли мне разбивать SQL-запрос на множественную строку? Я могу сделать это здесь, но это будет очень громоздко для большого SQL-запроса?

3. могу ли я преобразовать строку sql в API фрейма данных, используя какую-либо библиотеку, потому что ее будет очень сложно преобразовать вручную