Manipulando consultas grandes em fluxos de trabalho interativos

Um desafio com fluxos de trabalho de dados interativos é lidar com consultas grandes. Isso inclui consultas que geram muitas linhas de saída, buscam muitas partições externas ou computação em conjuntos de dados extremamente grandes. Essas consultas podem ser extremamente lentas, saturar recursos de computação e dificultar o compartilhamento da mesma computação por outras pessoas.

O Query Watchdog é um processo que impede que consultas monopolizem recursos de computação examinando as causas mais comuns de consultas grandes e encerrando consultas que ultrapassam um limite. Este artigo descreve como habilitar e configurar o Monitor de Consultas.

Um analista está executando algumas consultas ad hoc em um data warehouse just-in-time. O analista usa uma computação de dimensionamento automático compartilhado que facilita o uso de uma única computação por vários usuários ao mesmo tempo. Suponha que haja duas tabelas que cada uma tenha um milhão de linhas.

import org.apache.spark.sql.functions._
spark.conf.set("spark.sql.shuffle.partitions", 10)

spark.range(1000000)
  .withColumn("join_key", lit(" "))
  .createOrReplaceTempView("table_x")
spark.range(1000000)
  .withColumn("join_key", lit(" "))
  .createOrReplaceTempView("table_y")

Esses tamanhos de tabela são gerenciáveis no Apache Spark. No entanto, cada uma delas inclui uma coluna join_key com uma cadeia de caracteres vazia em cada linha. Isso pode acontecer se os dados não estiverem perfeitamente limpos ou se houver uma distorção de dados significativa em que algumas chaves são mais prevalentes do que outras. Essas chaves de junção vazias são muito mais prevalentes do que qualquer outro valor.

No código a seguir, o analista está unindo essas duas tabelas às respectivas chaves, o que produz a saída de um trilhão de resultados e todos eles são produzidos em um só executor (o executor que obtém a chave " "):

SELECT
  id, count(id)
FROM
  (SELECT
    x.id
  FROM
    table_x x
  JOIN
    table_y y
  on x.join_key = y.join_key)
GROUP BY id

Essa consulta parece estar em execução. Mas sem saber sobre os dados, o analista vê que há "apenas" uma única tarefa restante ao longo da execução do trabalho. A consulta nunca termina, deixando o analista frustrado e confuso sobre por que não funcionou.

Nesse caso, há apenas uma chave de junção problemática. Outras vezes pode haver muito mais.

Para habilitar e configurar o Query Watchdog, siga as etapas a seguir.

• Habilitar o Watchdog com spark.databricks.queryWatchdog.enabled.
• Configure o runtime da tarefa com spark.databricks.queryWatchdog.minTimeSecs.
• Exiba a saída com spark.databricks.queryWatchdog.minOutputRows.
• Configure a taxa de saída com spark.databricks.queryWatchdog.outputRatioThreshold.

Para impedir que uma consulta crie muitas linhas de saída para o número de linhas de entrada, você pode habilitar o Watchdog de Consulta e configurar o número máximo de linhas de saída como um múltiplo do número de linhas de entrada. Neste exemplo, usamos uma taxa de 1000 (o padrão).

spark.conf.set("spark.databricks.queryWatchdog.enabled", true)
spark.conf.set("spark.databricks.queryWatchdog.outputRatioThreshold", 1000L)

A última configuração declara que qualquer tarefa determinada nunca deve produzir mais de 1000 vezes o número de linhas de entrada.

Além de impedir que os usuários monopolizem recursos de computação para trabalhos que nunca serão concluídos, ele também economiza tempo ao falhar rapidamente em uma consulta que nunca teria sido concluída. Por exemplo, a consulta a seguir falhará após vários minutos porque excederá a taxa.

SELECT
  z.id
  join_key,
  sum(z.id),
  count(z.id)
FROM
  (SELECT
    x.id,
    y.join_key
  FROM
    table_x x
  JOIN
    table_y y
  on x.join_key = y.join_key) z
GROUP BY join_key, z.id

Veja o que você verá:

Geralmente, é suficiente habilitar o Query Watchdog e definir a relação entre saída e entrada, mas você também tem a opção de definir duas propriedades adicionais: spark.databricks.queryWatchdog.minTimeSecs e spark.databricks.queryWatchdog.minOutputRows. Essas propriedades especificam o tempo mínimo que uma determinada tarefa em uma consulta deve ser executada antes de cancelá-la e o número mínimo de linhas de saída para uma tarefa nessa consulta.

Por exemplo, você pode definir minTimeSecs como um valor mais alto se quiser dar a ele a chance de produzir um grande número de linhas por tarefa. Da mesma forma, você pode definir spark.databricks.queryWatchdog.minOutputRows como dez milhões se quiser interromper uma consulta somente após uma tarefa nessa consulta ter produzido dez milhões de linhas. Caso seja menor, a consulta será bem-sucedida, mesmo que a taxa de saída/entrada tenha sido excedida.

spark.conf.set("spark.databricks.queryWatchdog.minTimeSecs", 10L)
spark.conf.set("spark.databricks.queryWatchdog.minOutputRows", 100000L)

Outra consulta grande típica pode verificar uma grande quantidade de dados de grandes tabelas/conjuntos de dados. A operação de verificação pode durar muito tempo e saturar recursos de computação (até mesmo ler metadados de uma tabela do Hive grande pode levar muito tempo). Você pode definir maxHivePartitions para impedir a busca de muitas partições de uma tabela grande do Hive. Da mesma forma, você também pode definir maxQueryTasks para limitar consultas em um conjunto de dados extremamente grande.

spark.conf.set("spark.databricks.queryWatchdog.maxHivePartitions", 20000)
spark.conf.set("spark.databricks.queryWatchdog.maxQueryTasks", 20000)

O Watchdog de Consulta deve ser habilitado para a computação de análise ad hoc, em que analistas de SQL e cientistas de dados compartilham determinada computação e um administrador precisa garantir que as consultas sejam “bem executadas” entre si.

Em geral, não aconselhamos cancelar ansiosamente as consultas usadas em um cenário de ETL porque normalmente não há um humano no loop para corrigir o erro. Recomendamos que você desabilite o Query Watchdog para todos os casos, exceto para computação de análise ad hoc.