我使用以下代码将数据填充到Bigtable中:
CloudBigtableScanConfiguration config = new CloudBigtableScanConfiguration.Builder()
.withConfiguration("clusterId", options.getBigTableClusterId())
.withProjectId(options.getProject())
.withInstanceId(options.getBigTableInstanceId())
.withTableId(options.getOutputBTTable())
.build();
Pipeline p = Pipeline.create(options);
/**
* Read Data from Big Query
*/
CloudBigtableIO.initializeForWrite(p);
p.apply(BigQueryIO.Read.fromQuery(getQuery(options.getDate())))
.apply(ParDo.of(new DoFn<TableRow, Mutation>() {
public void processElement(ProcessContext c) {
Mutation output = convertDataToRow(c.element());
if (output != null) {
c.output(output);
};
}
}))
.apply(CloudBigtableIO.writeToTable(config));
p.run();
private static Mutation convertDataToRow(TableRow element) {
LOG.info("element: "+ element);
if(element.get("BASM_AID") != null){
Put obj = new Put(getRowKey(element).getBytes()).addColumn(SEGMENT_FAMILY, SEGMENT_COLUMN_NAME, ((String)element.get("BAS_category")).getBytes() );
obj.addColumn(USER_FAMILY, AID, ((String)element.get("BASM_AID")).getBytes());
if(element.get("BASM_segment_id") != null){
obj.addColumn(SEGMENT_FAMILY, SEGMENT_ID, ((String)element.get("BASM_segment_id")).getBytes());
}
if(element.get("BAS_sub_category") != null){
obj.addColumn(SEGMENT_FAMILY, SUB_CATEGORY, ((String)element.get("BAS_sub_category")).getBytes());
}
if(element.get("BAS_name") != null){
obj.addColumn(SEGMENT_FAMILY, NAME, ((String)element.get("BAS_name")).getBytes());
}
if(element.get("BAS_description") != null){
obj.addColumn(SEGMENT_FAMILY, DESCRIPTION, ((String)element.get("BAS_description")).getBytes());
}
if(element.get("BASM_krux_user_id") != null){
obj.addColumn(USER_FAMILY, KRUX_USER_ID, ((String)element.get("BASM_krux_user_id")).getBytes());
}
if(element.get("BAS_last_compute_day") != null){
obj.addColumn(SEGMENT_FAMILY, LAST_COMPUTE_DAY, ((String)element.get("BAS_last_compute_day")).getBytes());
}
if(element.get("BAS_type") != null){
obj.addColumn(SEGMENT_FAMILY, TYPE, ((String)element.get("BAS_type")).getBytes());
}
if(element.get("BASM_REGID") != null){
obj.addColumn(USER_FAMILY, REGID, ((String)element.get("BASM_REGID")).getBytes() );
}
return obj;
}else{
return null;
}
}我们有30个Bigtable节点,我的数据流作业是与100个工作人员一起工作,整个过程必须处理大约100亿行数据,使用上面的配置,我的作业需要一天多的时间才能完成,这是不理想的。
我知道增加Bigtable节点的数量是其中一个选择,但目前,我正在寻找其他不需要增加节点的选项。
发布于 2016-08-27 04:05:30
您可能希望看看this question。在此基础上,您需要查看写入带宽,如果超过80%,您可能希望减少工作节点的数量,或者要求增加配额并增加集群的大小。
发布于 2016-09-03 06:36:46
将大量数据大容量加载到空表中(无论是Bigtable还是HBase)将会出现性能问题,除非您提前对表进行了预拆分,因为最初没有任何平板电脑,因此所有写入都会流向单个服务器节点,分布不均匀。
因此,无论集群大小如何,初始大容量加载都不会获得高性能,因为它不会被并行化。
要解决此问题,您需要创建一个带有预拆分的表。您可以查看如何使用create tables in Cloud Bigtable via the HBase shell with pre-splits的示例。当我们运行一个基准测试将数据加载到Bigtable和HBase,we also pre-split the tables。
在具有大量数据的稳定状态下写入现有表时,将有许多平板电脑,这些平板电脑在集群上分布良好,因此性能会很好。但是,如果对空表执行大容量装载,则必须对其进行预拆分。
https://stackoverflow.com/questions/39169327
复制相似问题