spark hive job - renaming S3 dirs is very slow

[Сабина]

Запустила spark job для экспорта данных redshift -> hive . 100 mln records заняло аж 15 часов.
Копнула логи, а там все выполнено за 15 минут, а остальное время идет s3 directories renaming ( way how hive works from what I understand)

19362: INFO [2017-11-20 20:08:31,998] ({pool-2-thread-27} Hive.java[moveFile]:2642) - Renaming src: s3://..../.hive-staging_hive_2017-11-20_19-48-35_181_1682270180271573182-3/..../part-00005-db748631-e223-4d02-a548-5d6b6890ea3c.c000, dest: s3://...../part-00005-db748631-e223-4d02-a548-5d6b6890ea3c.c000, Status:true

Непонятно почему это все выполняется одним тредом все это время ?
Так задумано или что не подкручено в EMR кластере/Hive/Spark .... ?

[iDesperado]

думаю начать стоит с выяснения, что такое хайв ? что за енжин под хайвом ? там parquret файлики, hbase, что-то другое под низом ?
мы обычно пишем в parquet файлики, а потом командой CREATE TABLE .. STORED AS PARQUET LOCATION уже в хайф метастор что либо прописываем.

[Сабина]

думаю начать стоит с выяснения, что такое хайв ? что за енжин под хайвом ? там parquret файлики, hbase, что-то другое под низом ?
мы обычно пишем в parquet файлики, а потом командой CREATE TABLE .. STORED AS PARQUET LOCATION уже в хайф метастор что либо прописываем.

О! Я как раз так делала сначала, записывала из redshifт сразу в parquet но меня развернули
Там export из redshift делается делается в temp dir в tsv first, parquet тока уже когда попадате в hive.
PS. Нам надо в Hive external table

[Сабина]

OK, apparently there was some S3 issue which seem to go away now

[iDesperado]

а чем-то мотивировали двойную писанину? кстати, все таки интересно, что у EMR за енжин под хайвом ? MR или что посовременней ?

[_reality]

https://stackoverflow.com/a/42834182

[Сабина]

а чем-то мотивировали двойную писанину? кстати, все таки интересно, что у EMR за енжин под хайвом ? MR или что посовременней ?

spark-redshift драйвер так работает что если ты из редшифтовой таблицы делаешь dataframe, то он создает tempDir for unload, а нашем случае в S3 и по умолчанию tsv


обчественность грузит примерно так и меня заставили это использовать и запускать from notebook

Code: Select all

def loadDataFromRedshift(redshift_table: String,
                         select_query: String,
                         target_hive_table: String
                         ):Unit = {
    // Instance Profile for authentication to AWS resources
    val provider = new InstanceProfileCredentialsProvider()
    val credentials: AWSSessionCredentials = provider.getCredentials.asInstanceOf[AWSSessionCredentials]
    val token = credentials.getSessionToken
    val awsAccessKey = credentials.getAWSAccessKeyId
    val awsSecretKey = credentials.getAWSSecretKey
    
    //Provide the jdbc url for Amazon Redshift
    val jdbcUsername = "..."
    val jdbcPassword = "..."
    val jdbcHostname = "..."
    val jdbcPort = 5439
    val jdbcDatabase = "....."
    val jdbcUrl = s"jdbc:redshift://${jdbcHostname}:${jdbcPort}/${jdbcDatabase}?user=${jdbcUsername}&password=${jdbcPassword}"
    
    // Create and declare an S3 bucket where the temporary files are written
    val protocol  = "s3n"
    val s3Bucket  = "..."
    val s3TempDir = s"$protocol://$s3Bucket/..../$redshift_table"
    
    val df = spark.read
      .format("com.databricks.spark.redshift")
      .option("url", jdbcUrl)
          .option("tempdir", s3TempDir)
      .option("query", select_query)
      .option("temporary_aws_access_key_id", awsAccessKey)
      .option("temporary_aws_secret_access_key", awsSecretKey)
      .option("temporary_aws_session_token", token)
      .load()
    
    val sourceDF = df.drop("auto_id")
    val tmpTable: String = createOrReplaceTempView(target_hive_table, sourceDF)

    val partitionColumns = List("collection_date")
    val outputColumns = createSelectColumnsStatement(df.columns, spark.table(target_hive_table).columns, partitionColumns).mkString(", ")

    val sqlStatement =
      if (partitionColumns.isEmpty) {
        s"""
           |INSERT INTO TABLE $target_hive_table
           |SELECT $outputColumns
           |FROM $tmpTable
    """.stripMargin
      } else {
        val partitionFieldsStr = partitionColumns.mkString(", ")
        s"""
           |INSERT INTO TABLE $target_hive_table PARTITION($partitionFieldsStr)
           |SELECT $outputColumns, to_date(`collection_timestamp`) AS `collection_dt`
           |FROM $tmpTable
    """.stripMargin
      }
    
    println(s"SQL Insert statement: $sqlStatement")
    
    spark.sql(sqlStatement)
} 

Сама я до этого запускала from laptop

Code: Select all

def importRedshiftResultSetToHive(resultSetQuery: String, hiveTable: String, s3TempDir: String) {

    val redshiftDf = spark.read
      .format("com.databricks.spark.redshift")
      .option("url", jdbcURL)
      .option("query", resultSetQuery)
      .option("tempdir", s3TempDir)
      .option("forward_spark_s3_credentials", true)
      .load()

    //redshiftDf.show()

    import spark.implicits._

    val completeDf = redshiftDf
      .drop("auto_id")
      .withColumn("collection_dt", to_date($"collection_timestamp", "yyyy-MM-dd"))
      .withColumn("collection_timestamp", to_unix_timestamp($"collection_timestamp"))
      .withColumn("event_timestamp", to_unix_timestamp($"event_timestamp"))
      .withColumn("etl_timestamp", to_unix_timestamp($"etl_timestamp"))

    //completeDf.show()
    completeDf.write.format("parquet").insertInto(hiveTable)
  }

Но оно так же работает rename etc

[Сабина]

https://stackoverflow.com/a/42834182

Интересно
Наш девопсовый перец клянется что все что он поменял - это tez.runtime.pipeline.sorter.lazy-allocate.memory на true и потом перестартанул все сервисы оно "залетало". У нас Tez= hive engine
Я не поняла каким местом этот параметр мог помочь, наверное иррелевант. Скорее там какая то сопля была и перезапуск всех сервисов помог.
У меня последний день тут, так уж умничаю напоследок , ну и вообще люблю разбираться с performance до конца. Но тут не получилось - трудно без пермишенсов да еще и когда кто-то посторонний меняет и перестартует фиг знает что

[metaller]

https://stackoverflow.com/a/42834182

Интересно
Наш девопсовый перец клянется что все что он поменял - это tez.runtime.pipeline.sorter.lazy-allocate.memory на true и потом перестартанул все сервисы оно "залетало". У нас Tez= hive engine
Я не поняла каким местом этот параметр мог помочь, наверное иррелевант. Скорее там какая то сопля была и перезапуск всех сервисов помог.
У меня последний день тут, так уж умничаю напоследок , ну и вообще люблю разбираться с performance до конца. Но тут не получилось - трудно без пермишенсов да еще и когда кто-то посторонний меняет и перестартует фиг знает что

Попробуйте без TEZ, может помочь. За мою недолгую (1.5 года) карьеру Big Data developer слово TEZ стало синонимом слова ГЛЮК

[iDesperado]

с налету не нашел описания как работает completeDf.write.format("parquet").insertInto(hiveTable)
вдруг оно SQL в хайв отправляет, а не то что мы подумали ?

у меня другой вариант был: completeDf.write.parquet("folder") а потом уже на фолдер надевать хайв таблички

[Сабина]

Попробуйте без TEZ, может помочь. За мою недолгую (1.5 года) карьеру Big Data developer слово TEZ стало синонимом слова ГЛЮК

Ой на больной мозоль наступаете ! В смысле что Tez притащил тот самый девопсовый гай, который "лучше девелоперов знает что делать". Ну и сами понимаете стартап, то есть с кем работать тим не выбирает - кого дали, то бишь.

Меня еще одна вещь мучает ( точнее несколько , но теперь уже задним числом).
Там кластер был сконфигурирован так:
- MASTER node,
- CORE node
- TASK node ( автоскейлится от 1 до 10).

Смущали следующие вещи:
1) Zeppelin был сконфигурирован так, что все ,запущенное в разных ноутбуках (разными девелоперами, пользующимися одним общим логином) проходило как один application ID. Логи трудно разгребать, все в куче. По моему это не самый эффективный способ конфигурировать notebook for multiple developers
2) В DAG видно что для некоторых задач запускаются десятки экзекьюторов, но все они почему то на CORE node, то есть IP core node, только порты разные. А почему TASK node(s) вообще не используется ? Это тоже не поняла ?

А почему ваша карьера big data engineer недолгая ? забросили , непонравилась ? или только начинаете ?

[Сабина]

с налету не нашел описания как работает completeDf.write.format("parquet").insertInto(hiveTable)
вдруг оно SQL в хайв отправляет, а не то что мы подумали ?

у меня другой вариант был: completeDf.write.parquet("folder") а потом уже на фолдер надевать хайв таблички

как это sql отправляет ? Отправляло что надо насколько я видела .
в вашем случае как я поняла вы писали не в external table ... , в нашем мы сначала создавали CREATE AS EXTERNAL TABLe с нужными партициями, а потом в них экспортировали данные из редшифта

[metaller]

Попробуйте без TEZ, может помочь. За мою недолгую (1.5 года) карьеру Big Data developer слово TEZ стало синонимом слова ГЛЮК

Ой на больной мозоль наступаете ! В смысле что Tez притащил тот самый девопсовый гай, который "лучше девелоперов знает что делать". Ну и сами понимаете стартап, то есть с кем работать тим не выбирает - кого дали, то бишь.

Меня еще одна вещь мучает ( точнее несколько , но теперь уже задним числом).
Там кластер был сконфигурирован так:
- MASTER node,
- CORE node
- TASK node ( автоскейлится от 1 до 10).

Смущали следующие вещи:
1) Zeppelin был сконфигурирован так, что все ,запущенное в разных ноутбуках (разными девелоперами, пользующимися одним общим логином) проходило как один application ID. Логи трудно разгребать, все в куче. По моему это не самый эффективный способ конфигурировать notebook for multiple developers
2) В DAG видно что для некоторых задач запускаются десятки экзекьюторов, но все они почему то на CORE node, то есть IP core node, только порты разные. А почему TASK node(s) вообще не используется ? Это тоже не поняла ?

А почему ваша карьера big data engineer недолгая ? забросили , непонравилась ? или только начинаете ?

Потому что к тому времени получил GC и решил сменить род деятельности на не-IT. 20 лет по клаве стучал, хватит уже

[Сабина]

Потому что к тому времени получил GC и решил сменить род деятельности на не-IT. 20 лет по клаве стучал, хватит уже

А! Ну да. Память короткая