- android - RelativeLayout 背景可绘制重叠内容
- android - 如何链接 cpufeatures lib 以获取 native android 库?
- java - OnItemClickListener 不起作用,但 OnLongItemClickListener 在自定义 ListView 中起作用
- java - Android 文件转字符串
我尝试从 HDFS 中的文件聚合数据。我需要从那些对 hbase 中的特定表具有值(value)的数据中添加一些详细信息。
但我有异常(exception):
org.apache.spark.SparkException: Task not serializable
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:166)
at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:158)
at org.apache.spark.SparkContext.clean(SparkContext.scala:1623)
at org.apache.spark.rdd.RDD.map(RDD.scala:286)
at org.apache.spark.api.java.JavaRDDLike$class.mapToPair(JavaRDDLike.scala:113)
at org.apache.spark.api.java.AbstractJavaRDDLike.mapToPair(JavaRDDLike.scala:46)
at ......
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:497)
at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:577)
at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:174)
at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:197)
at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:112)
at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
Caused by: java.io.NotSerializableException: org.apache.hadoop.hbase.client.ConnectionManager$HConnectionImplementation
Serialization stack:
at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:38)
at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:47)
at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:80)
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:164)
我知道问题发生在我们尝试在 map 函数期间访问 hbase 时。
我的问题是:如何使用 hbase 表中包含的值来完成我的 RDD。
例如:hdfs 中的文件是 csv:
Name;Number1;Number2
toto;1;2
在 hbase 中,我们有与名称 toto 关联的数据。
我需要检索数字 1 和数字 2 的总和(这是最简单的部分)并与表中的数据聚合。例如:
reducer 的键将是 tata 并通过获取 hbase 表中的行键 toto 来检索。
有什么建议吗?
最佳答案
最后一位同事做到了,多亏了你的建议:
所以这是允许使用 hbase 表中的数据聚合文件的映射代码。
private final Logger LOGGER = LoggerFactory.getLogger(AbtractGetSDMapFunction.class);
/**
* Namespace name
*/
public static final String NAMESPACE = "NameSpace";
private static final String ID = "id";
private Connection connection = null;
private static final String LINEID = "l";
private static final String CHANGE_LINE_ID = "clid";
private static final String CHANGE_LINE_DATE = "cld";
private String constClientPortHBase;
private String constQuorumHBase;
private int constTimeOutHBase;
private String constZnodeHBase;
public void initConnection() {
Configuration conf = HBaseConfiguration.create();
conf.setInt("timeout", constTimeOutHBase);
conf.set("hbase.zookeeper.quorum", constQuorumHBase);
conf.set("hbase.zookeeper.property.clientPort", constClientPortHBase);
conf.set("zookeeper.znode.parent", constZnodeHBase);
try {
connection = HConnectionManager.createConnection(conf);
} catch (Exception e) {
LOGGER.error("Error in the configuration of the connection with HBase.", e);
}
}
public Tuple2<String, myInput> call(String row) throws Exception {
//this is where you need to init the connection for hbase to avoid serialization problem
initConnection();
....do your work
State state = getCurrentState(myInput.getKey());
....do your work
}
public AbtractGetSDMapFunction( String constClientPortHBase, String constQuorumHBase, String constZnodeHBase, int constTimeOutHBase) {
this.constClientPortHBase = constClientPortHBase;
this.constQuorumHBase = constQuorumHBase;
this.constZnodeHBase = constZnodeHBase;
this.constTimeOutHBase = constTimeOutHBase;
}
/***************************************************************************/
/**
* Table Name
*/
public static final String TABLE_NAME = "Table";
public state getCurrentState(String key) throws TechnicalException {
LOGGER.debug("start key {}", key);
String buildRowKey = buildRowKey(key);
State currentState = new State();
String columnFamily = State.getColumnFamily();
if (!StringUtils.isEmpty(buildRowKey) && null != columnFamily) {
try {
Get scan = new Get(Bytes.toBytes(buildRowKey));
scan.addFamily(Bytes.toBytes(columnFamily));
addColumnsToScan(scan, columnFamily, ID);
Result result = getTable().get(scan);
currentState.setCurrentId(getLong(result, columnFamily, ID));
} catch (IOException ex) {
throw new TechnicalException(ex);
}
LOGGER.debug("end ");
}
return currentState;
}
/***********************************************************/
private Table getTable() throws IOException, TechnicalException {
Connection connection = getConnection();
// Table retrieve
if (connection != null) {
Table table = connection.getTable(TableName.valueOf(NAMESPACE, TABLE_NAME));
return table;
} else {
throw new TechnicalException("Connection to Hbase not available");
}
}
/****************************************************************/
private Long getLong(Result result, String columnFamily, String qualifier) {
Long toLong = null;
if (null != columnFamily && null != qualifier) {
byte[] value = result.getValue(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier));
toLong = (value != null ? Bytes.toLong(value) : null);
}
return toLong;
}
private String getString(Result result, String columnFamily, String qualifier) {
String toString = null;
if (null != columnFamily && null != qualifier) {
byte[] value = result.getValue(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier));
toString = (value != null ? Bytes.toString(value) : null);
}
return toString;
}
public Connection getConnection() {
return connection;
}
public void setConnection(Connection connection) {
this.connection = connection;
}
private void addColumnsToScan(Get scan, String family, String qualifier) {
if (org.apache.commons.lang.StringUtils.isNotEmpty(family) && org.apache.commons.lang.StringUtils.isNotEmpty(qualifier)) {
scan.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
}
}
private String buildRowKey(String key) throws TechnicalException {
StringBuilder rowKeyBuilder = new StringBuilder();
rowKeyBuilder.append(HashFunction.makeSHA1Hash(key));
return rowKeyBuilder.toString();
}
关于java - 在 map 调用中获取 Spark 上的行,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/41985552/
我正在尝试从一个 map 的 map 的 map 的 map 的 map 的 map 的 map 的 map 的 map 的 map 的 map 的 map 的 map 的 map 的 map 的 m
我是 Haskell 的新手,我认为函数 map map和 map.map在 Haskell 中是一样的。 我的终端给了我两种不同的类型, (map.map) :: (a -> b) -> [[a]
我的目标是创建一个 map 的 map ,这样我就可以通过它的键检索外部 map 的信息,然后通过它们的键访问它的“内部” map 。 但是,当我得到每个内部映射时,我最初创建的映射变成了一个对象,我
如何使用 Java8 编写以下代码? for (Entry> entry : data.entrySet()) { Map value = entry.getValue(); if (valu
我有覆盖整个南非的图片。它们为Tiff格式,并已将坐标嵌入其中。我正在尝试拍摄这些图像(大约20张图像),并将它们用作我的iPhone应用程序中的地图叠加层。我的问题在于(准确地)将地图切成图块。 我
所以我有 2 std::map s >一个是“旧的”,一个是“新的”,我想知道哪些文件被删除了,这样就能够遍历差异并对 shared_ptr 做一些事情。这样的事情可能吗?如何做到? 最佳答案 虽然
是否可以将当前查看的 google.maps.Map 转换为静态图像链接,以便我可以获取图像并将其嵌入到 PDF 中? 我在 map 上添加了一些带有自定义图标的标记,所以我不确定这是否真的可行。 如
你能帮我吗 Java Streams ? 从标题可以看出我需要合并List>>进入Map> . 列表表示为List>>看起来像: [ { "USER_1":{
对于 idAndTags 的第二个条目,内部映射被打乱,但第一个条目则不然 第一次接近! for (Map.Entry> entryOne : idAndTags.entrySet()) {
我将从我的代码开始,因为它应该更容易理解我想要做什么: @function get-color($color, $lightness) { @return map-get(map-get($col
我过去曾在许多网站上使用过 Google map ,但遇到了以前从未遇到过的问题。 map 窗口正在显示,但它只显示左上角的 map 片段,以及之后的任何内容(即使我在周围导航时),右侧也不会加载任何
众所周知,这些 map ,无论是常规街道 map 还是卫星 map ,在中国的特定地区都无法正确排列。那么哪个 map 排列正确,是卫星 map 还是默认街道 map ?一些网站表明卫星 map 是正
在拖尾事件之后,我面临着获取此处 map 中的 map 边界的问题。我需要新的经纬度来在新更改的视口(viewport)中获取一些项目/点。我只是想在拖动结束时获得谷歌地图map.getBounds(
我想做的是通过 ajax API 显示以英国邮政编码为中心的小型 bing 生成 map 。我相信这是可能的;我在 Bing map 文档中找不到如何将英国邮政编码转换为可以插入 map Ajax 控
我有一个 List我想转换成的 e Map>其中外部字符串应为“Name”,内部字符串应为“Domain”。 Name Id Domain e(0) - Emp1, 1, Insuran
我的第 2 部分:https://stackoverflow.com/questions/21780627/c-map-of-maps-typedef-doubts-queries 然后我继续创建 I
是否可以在 1 行中使用 Java8 编写以下所有 null 和空字符串检查? Map> data = new HashMap<>(holdings.rowMap()); Set>> entrySet
我正在审查一个项目的旧代码,并使用 Map 的 Map 的 Map 获得了如下数据结构(3 层 map ): // data structure Map>>> tagTree
这可能是一种不好的做法,但我还没有找到更好的解决方案来解决我的问题。所以我有这张 map // Map>> private Map>> properties; 我想初始化它,这样我就不会得到 Null
我们在 JDK 1.7 中使用 HashMap,我在使用 SonarQube 进行代码审查时遇到了一些问题。 请考虑以下示例: public class SerializationTest imple
我是一名优秀的程序员,十分优秀!