Массовая вставка данных в HBase с использованием структурированной потоковой передачи Spark
Я читаю данные, поступающие от Kafka (100 000 строк в секунду) с использованием Structured Spark Streaming, и пытаюсь вставить все данные в HBase.
Я использую Cloudera Hadoop 2.6 и Spark 2.3.
Я пробовал что-то вроде того, что видел здесь.
eventhubs.writeStream
.foreach(new MyHBaseWriter[Row])
.option("checkpointLocation", checkpointDir)
.start()
.awaitTermination()
MyHBaseWriter выглядит так:
class AtomeHBaseWriter[RECORD] extends HBaseForeachWriter[Row] {
override def toPut(record: Row): Put = {
override val tableName: String = "hbase-table-name"
override def toPut(record: Row): Put = {
// Get Json
val data = JSON.parseFull(record.getString(0)).asInstanceOf[Some[Map[String, Object]]]
val key = data.getOrElse(Map())("key")+ ""
val val = data.getOrElse(Map())("val")+ ""
val p = new Put(Bytes.toBytes(key))
//Add columns ...
p.addColumn(Bytes.toBytes(columnFamaliyName),Bytes.toBytes(columnName), Bytes.toBytes(val))
p
}
}
И класс HBaseForeachWriter выглядит так:
trait HBaseForeachWriter[RECORD] extends ForeachWriter[RECORD] {
val tableName: String
def pool: Option[ExecutorService] = None
def user: Option[User] = None
private var hTable: Table = _
private var connection: Connection = _
override def open(partitionId: Long, version: Long): Boolean = {
connection = createConnection()
hTable = getHTable(connection)
true
}
def createConnection(): Connection = {
// I create HBase Connection Here
}
def getHTable(connection: Connection): Table = {
connection.getTable(TableName.valueOf(Variables.getTableName()))
}
override def process(record: RECORD): Unit = {
val put = toPut(record)
hTable.put(put)
}
override def close(errorOrNull: Throwable): Unit = {
hTable.close()
connection.close()
}
def toPut(record: RECORD): Put
}
Итак, здесь я делаю пут построчно, даже если я разрешаю 20 исполнителей и 4 ядра для каждого, у меня нет данных, вставленных сразу в HBase. Итак, что мне нужно сделать, это массовая загрузка, но я борюсь, потому что все, что я нахожу в Интернете, - это реализовать это с помощью RDD и Map/Reduce.
Ответы 1
Насколько я понимаю, это низкая скорость загрузки записей в hbase. У меня есть несколько предложений к вам.
1) hbase.client.write.buffer .
приведенное ниже свойство может вам помочь.
hbase.client.write.bufferDescription Default size of the BufferedMutator write buffer in bytes. A bigger buffer takes more memory — on both the client and server side since server instantiates the passed write buffer to process it — but a larger buffer size reduces the number of RPCs made. For an estimate of server-side memory-used, evaluate hbase.client.write.buffer * hbase.regionserver.handler.count
Default 2097152 (around 2 mb )
Я предпочитаю foreachBatch см. искровые документы (своего рода foreachPartition в ядре spark) foreach
Также в вашем hbase Writer расширяется ForeachWriter
open метод инициализации списка массивов
в process добавьте пут в массив путов
в closetable.put(listofputs);, а затем сбросьте список массивов после обновления таблицы...
то, что он делает, в основном, ваш размер буфера, упомянутый выше, заполнен 2 МБ, после чего он будет сброшен в таблицу hbase. до этого записи не попадут в таблицу hbase.
вы можете увеличить это до 10 МБ и так далее.... Таким образом количество RPC будет уменьшено. и огромный кусок данных будет сброшен и окажется в таблице hbase.
когда буфер записи заполняется и запускается таблица flushCommits in to hbase.
Пример кода: в моем отвечать
2) выключить WAL вы можете отключить WAL (упреждающая запись в журнал - Опасно отсутствие восстановления), но это ускорит запись... если вы не хотите восстанавливать данные.
Note : if you are using solr or cloudera search on hbase tables you should not turn it off since Solr will work on WAL. if you switch it off then, Solr indexing wont work.. this is one common mistake many of us does.
Как отключить:https://hbase.apache.org/1.1/apidocs/org/apache/hadoop/hbase/client/Put.html#setWriteToWAL (логическое значение)
Базовая архитектура и связь для дальнейшего изучения:
как я уже упоминал, список путов - это хороший способ ... это старый способ (foreachPartition со списком путов) сделать до того, как пример структурированной потоковой передачи будет показан ниже ... где foreachPartition работает для каждого раздела, а не для каждой строки.
def writeHbase(mydataframe: DataFrame) = {
val columnFamilyName: String = "c"
mydataframe.foreachPartition(rows => {
val puts = new util.ArrayList[ Put ]
rows.foreach(row => {
val key = row.getAs[ String ]("rowKey")
val p = new Put(Bytes.toBytes(key))
val columnV = row.getAs[ Double ]("x")
val columnT = row.getAs[ Long ]("y")
p.addColumn(
Bytes.toBytes(columnFamilyName),
Bytes.toBytes("x"),
Bytes.toBytes(columnX)
)
p.addColumn(
Bytes.toBytes(columnFamilyName),
Bytes.toBytes("y"),
Bytes.toBytes(columnY)
)
puts.add(p)
})
HBaseUtil.putRows(hbaseZookeeperQuorum, hbaseTableName, puts)
})
}
To sumup :
What I feel is we need to understand the psycology of spark and hbase to make then effective pair.
Хорошо, тогда попробуйте список вложений в foreach, который должен работать.
Привет, Рам, я обновил свою версию до 2.4. Где я могу найти класс HBaseUtil? Спасибо
Но foreachBatch доступен только с Spark 2.4, у меня 2.3 :(