Мы использовали кластер EMR для целей разработки, и там я использовал библиотеку ядра, предоставляемую hortornworks, для загрузки таблиц HBase непосредственно из фреймов данных spark, и все это работало нормально.
Недавно мы перешли на использованиеCloudera 5.14 (который по умолчанию поставляется с spark 1.6 и scala 2.10, но мы обновили его до версии 2.3 и scala 2.11).Загрузка hbase завершается с ошибкой ниже:
Exception in thread "main" java.lang.NoSuchMethodError: org.apache.hadoop.hbase.HTableDescriptor.addFamily(Lorg/apache/hadoop/hbase/HColumnDescriptor;)Lorg/apache/hadoop/hbase/HTableDescriptor;
at org.apache.spark.sql.execution.datasources.hbase.HBaseRelation$$anonfun$createTable$1.apply(HBaseRelation.scala:149)
at org.apache.spark.sql.execution.datasources.hbase.HBaseRelation$$anonfun$createTable$1.apply(HBaseRelation.scala:146)
at scala.collection.immutable.List.foreach(List.scala:381)
at org.apache.spark.sql.execution.datasources.hbase.HBaseRelation.createTable(HBaseRelation.scala:146)
at org.apache.spark.sql.execution.datasources.hbase.DefaultSource.createRelation(HBaseRelation.scala:60)
at org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand.run(SaveIntoDataSourceCommand.scala:46)
at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:70)
at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:68)
at org.apache.spark.sql.execution.command.ExecutedCommandExec.doExecute(commands.scala:86)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:131)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:127)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:155)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:152)
at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:127)
at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:80)
at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:80)
at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:654)
at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:654)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:77)
at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:654)
at org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:273)
at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:267)
at com.sc.eni.config.HbaseConfiguration$.hbaseload(HbaseConfiguration.scala:47)
at com.sc.eni.core.Ingest$.loadLayer4(Ingest.scala:102)
at com.sc.eni.core.Ingest$$anonfun$runPipeline$1.apply(Ingest.scala:23)
at com.sc.eni.core.Ingest$$anonfun$runPipeline$1.apply(Ingest.scala:21)
at scala.collection.immutable.List.map(List.scala:273)
at com.sc.eni.core.Ingest$.runPipeline(Ingest.scala:17)
at com.sc.eni.main.tssStart$.main(tssStart.scala:38)
at com.sc.eni.main.tssStart.main(tssStart.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43
Мы используем код / версии ->
cloudera verion: 1.2.0-cdh5.14.0
shc core: 1.1.1-2.1-s_2.11 (Репо: http://repo.hortonworks.com/content/repositories/releases/)
Искра: 2.3.0
Скала: 2.11.8
table catalogue : def catalog = s"""{
|"table":{"namespace":"default", "name":"test","tableCoder":"PrimitiveType"},
|"rowkey":"tradeid",
|"columns":{
|"tradeid":{"cf":"rowkey", "col":"tradeid", "type":"string"},
|"eoddate":{"cf":"td", "col":"eoddate", "type":"string"},
|"buysell":{"cf":"td", "col":"buysell", "type":"string"},
|"trader":{"cf":"td", "col":"trader", "type":"string"},
|"exclegalentity":{"cf":"td", "col":"exclegalentity", "type":"string"},
|"intaccountid":{"cf":"td", "col":"intaccountid", "type":"string"},
|"intaccbook":{"cf":"td", "col":"intaccbook", "type":"string"}
|}
|}""".stripMargin
df.write
.options(Map(HBaseTableCatalog.tableCatalog -> catalog,HBaseTableCatalog.newTable -> "5"))
.format("org.apache.spark.sql.execution.datasources.hbase").save()
Буду очень признателен за любую помощь.