Spark jar работает слишком долго [дубликат]

Подходит ли несериализованная таблица к 40 ГБ?

В любом случае, когда вы используете стандартный JDBC с синтаксисом ansi SQL, вы используете механизм БД, поэтому, если teradata (я не знаю teradata) содержит статистику вашего таблица, классический «select count (*) из таблицы» будет очень быстрым. Вместо этого искру загружает 100 миллионов строк в памяти с чем-то вроде «select * from table», а затем будет выполнять подсчет строк RDD. Это отличная рабочая нагрузка.

Все операции агрегации выполняются после того, как весь набор данных будет извлечен в память в коллекцию DataFrame. Таким образом, счет в Spark никогда не будет таким эффективным, как это было бы непосредственно в TeraData. Иногда стоит подталкивать некоторые вычисления в базу данных, создавая представления, а затем сопоставляя эти представления с помощью JDBC API.

Каждый раз, когда вы используете драйвер JDBC для доступа к большой таблице, вы должны указать стратегию разделения в противном случае вы создадите DataFrame / RDD с одним разделом, и вы перегрузите одно соединение JDBC.

Вместо этого вы хотите попробовать следующий AI (начиная с Spark 1.4.0 +):

sqlctx.read.jdbc(
  url = "<URL>",
  table = "<TABLE>",
  columnName = "<INTEGRAL_COLUMN_TO_PARTITION>", 
  lowerBound = minValue,
  upperBound = maxValue,
  numPartitions = 20,
  connectionProperties = new java.util.Properties()
)

Существует также возможность отбрасывать некоторую фильтрацию.

Если у вас нет равномерно распределенного интегрального столбца, вы хотите создать некоторые пользовательские разделы, указав пользовательские предикаты (where). Например, предположим, что у вас есть столбец timestamp и вы хотите разделить по диапазонам дат:

    val predicates = 
  Array(
    "2015-06-20" -> "2015-06-30",
    "2015-07-01" -> "2015-07-10",
    "2015-07-11" -> "2015-07-20",
    "2015-07-21" -> "2015-07-31"
  )
  .map {
    case (start, end) => 
      s"cast(DAT_TME as date) >= date '$start'  AND cast(DAT_TME as date) <= date '$end'"
  }

 predicates.foreach(println) 

// Below is the result of how predicates were formed 
//cast(DAT_TME as date) >= date '2015-06-20'  AND cast(DAT_TME as date) <= date '2015-06-30'
//cast(DAT_TME as date) >= date '2015-07-01'  AND cast(DAT_TME as date) <= date '2015-07-10'
//cast(DAT_TME as date) >= date '2015-07-11'  AND cast(DAT_TME as date) <= date //'2015-07-20'
//cast(DAT_TME as date) >= date '2015-07-21'  AND cast(DAT_TME as date) <= date '2015-07-31'


sqlctx.read.jdbc(
  url = "<URL>",
  table = "<TABLE>",
  predicates = predicates,
  connectionProperties = new java.util.Properties()
)

Он сгенерирует DataFrame, где каждый раздел будет содержать записи каждого подзапроса, связанные с различными предикатами.

Проверьте исходный код на DataFrameReader.scala