scala—创建hfiles时的spark问题—添加了一个在词汇上不比前一个单元格大的键

mqkwyuun 于 2021-06-09 发布在 Hbase

关注(0)|答案(1)|浏览(365)

我正在尝试创建hfiles来批量加载到hbase中，它会不断抛出带有row键的错误，即使一切看起来都很好。我正在使用以下代码：

val df = sqlContext.read.format("com.databricks.spark.csv")
   .option("header", "true")
   .option("inferSchema", "true")
   .load("data.csv")

import sqlContext.implicits._

val DF2 = df.filter($"company".isNotNull)
  .dropDuplicates(Array("company"))
  .sortWithinPartitions("company").sort("company")

val rdd = DF2.flatMap(x => { 
  val rowKey = Bytes.toBytes(x(0).toString)
  for (i <- 0 to cols.length - 1) yield {
    val index = x.fieldIndex(new String(cols(i)))
    val value = if (x.isNullAt(index)) "".getBytes else x(index).toString.getBytes
         (new ImmutableBytesWritable(rowKey), new KeyValue(rowKey, COLUMN_FAMILY, cols(i), value))
  }
})

rdd.saveAsNewAPIHadoopFile("HDFS LOcation", classOf[ImmutableBytesWritable], classOf[KeyValue], classOf[HFileOutputFormat2], fconf)

我使用以下数据

company,date,open,high,low,close,volume
ABG,01-Jan-2010,11.53,11.53,11.53,11.53,0
ABM,01-Jan-2010,20.66,20.66,20.66,20.66,0
ABR,01-Jan-2010,1.99,1.99,1.99,1.99,0
ABT,01-Jan-2010,53.99,53.99,53.99,53.99,0
ABX,01-Jan-2010,39.38,39.38,39.38,39.38,0
ACC,01-Jan-2010,28.1,28.1,28.1,28.1,0
ACE,01-Jan-2010,50.4,50.4,50.4,50.4,0
ACG,01-Jan-2010,8.25,8.25,8.25,8.25,0
ADC,01-Jan-2010,27.25,27.25,27.25,27.25,0

它抛出错误为

java.io.IOException: Added a key not lexically larger than previous. Current cell = ADC/data:high/1505862570671/Put/vlen=5/seqid=0, lastCell = ADC/data:open/1505862570671/Put/vlen=5/seqid=0
    at org.apache.hadoop.hbase.io.hfile.AbstractHFileWriter.checkKey(AbstractHFileWriter.java:204)
    at org.apache.hadoop.hbase.io.hfile.HFileWriterV2.append(HFileWriterV2.java:265)
    at org.apache.hadoop.hbase.regionserver.StoreFile$Writer.append(StoreFile.java:992)
    at org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2$1.write(HFileOutputFormat2.java:199)

我甚至尝试过对数据进行排序，但仍然抛出了错误。

hbase scala apache-spark bulk-load

来源：https://stackoverflow.com/questions/56022911/unable-to-insert-more-than-3-columns-in-a-column-family

1条答案

按热度按时间

qij5mzcb1#

我花了几个小时找到了解决方案，根本原因是列没有排序。
因为hfile需要按字典顺序排序的keyvalue，而且在编写时也需要按您的情况排序 HFileOutputFormat2->AbstractHFileWriter 找到 Added a key not lexically larger than previous. Current cell . 您已经在行级别应用了排序，一旦您对列进行了排序，它也会起作用。
这里的问题很好地解释了为什么hbase keyvaluesortreducer需要对所有keyvalue进行排序。
解决方案：

//sort columns
val cols = companyDs.columns.sorted

//Rest of the code is same

val output = companyDs.rdd.flatMap(x => {
  val rowKey = Bytes.toBytes(x(0).toString)
 val hkey = new ImmutableBytesWritable(rowKey)
  for (i <- 0 to cols.length - 1) yield {
    val index = x.fieldIndex(new String(cols(i)))
    val value = if (x.isNullAt(index)) "".getBytes else x(index).toString.getBytes
    val kv = new KeyValue(rowKey,COLUMN_FAMILY, cols(i).getBytes(),System.currentTimeMillis()+i ,x(i).toString.getBytes())
    (hkey,kv)
  }
})
output.saveAsNewAPIHadoopFile("<path>"
  , classOf[ImmutableBytesWritable], classOf[KeyValue],
  classOf[HFileOutputFormat2], config)

赞(0）回复(0）举报 2021-06-09

我来回答

scala—创建hfiles时的spark问题—添加了一个在词汇上不比前一个单元格大的键

1条答案

相关问题

热门标签

最新问答