使用hbase自定义筛选器时出现异常

hl0ma9xz  于 2021-05-29  发布在  Hadoop
关注(0)|答案(2)|浏览(515)

我已经编写了一个hbase自定义过滤器 FilterBase 变成了jar。过滤器如下所示：

public class MyFilter1 extends FilterBase implements Serializable{
boolean filterRow= true;
String srh;

public MyFilter1(String str) {
    this.srh= str;
}

@Override
public ReturnCode filterKeyValue(Cell c) throws IOException {
    String str= Bytes.toString(c.getValue());

    if(str.contains(str)) {
        filterRow= false;
        return ReturnCode.INCLUDE;
    }

    filterRow= true;
    return ReturnCode.SKIP;
}

@Override
public  boolean filterRow() {
    return filterRow;
}

@Override
public byte[] toByteArray() throws IOException {
    ByteArrayOutputStream out = new ByteArrayOutputStream();
    ObjectOutputStream os = new ObjectOutputStream(out);
    os.writeObject(this);
    return out.toByteArray();
}

public static MyFilter1 parseFrom(final byte[] data) {
    ByteArrayInputStream in = new ByteArrayInputStream(data);
    MyFilter1 ans= null;
    ObjectInputStream is;
    try {
        is = new ObjectInputStream(in);
        ans= (MyFilter1)is.readObject();;
    } catch (Exception e) {
        e.printStackTrace();
    }
    return ans;
}

}
在生成jar文件（即myfilter.jar）之后，我将它放在/use/local/hbase/lib/filters目录中。然后我开始
export hbase\u classpath=“/usr/local/hbase/lib/filters/myfilter.jar”
在hbase-env.sh中筛选并重新启动hbase服务器。然后我使用java程序中的自定义过滤器：

public static void main(String argv[]) throws IOException {
        Configuration conf= HBaseConfiguration.create();
        Connection con= ConnectionFactory.createConnection(conf);

        Table table= con.getTable(TableName.valueOf("stud"));

        Filter fl= new MyFilter("uc");

        Scan sc= new Scan();
        sc.setFilter(fl);

        ResultScanner rs= table.getScanner(sc);

        for(Result r : rs)
            System.out.println(Bytes.toString(r.getValue(Bytes.toBytes("perData"), Bytes.toBytes("name"))));
    }

但有以下例外

Exception in thread "main" org.apache.hadoop.hbase.DoNotRetryIOException: org.apache.hadoop.hbase.DoNotRetryIOException: java.lang.reflect.InvocationTargetException
at org.apache.hadoop.hbase.protobuf.ProtobufUtil.toFilter(ProtobufUtil.java:1478)
at org.apache.hadoop.hbase.protobuf.ProtobufUtil.toScan(ProtobufUtil.java:993)
at org.apache.hadoop.hbase.regionserver.RSRpcServices.scan(RSRpcServices.java:2396)
at org.apache.hadoop.hbase.protobuf.generated.ClientProtos$ClientService$2.callBlockingMethod(ClientProtos.java:33648)
at org.apache.hadoop.hbase.ipc.RpcServer.call(RpcServer.java:2180)
at org.apache.hadoop.hbase.ipc.CallRunner.run(CallRunner.java:112)
at org.apache.hadoop.hbase.ipc.RpcExecutor.consumerLoop(RpcExecutor.java:133)
at org.apache.hadoop.hbase.ipc.RpcExecutor$1.run(RpcExecutor.java:108)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.reflect.InvocationTargetException
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.hadoop.hbase.protobuf.ProtobufUtil.toFilter(ProtobufUtil.java:1474)
... 8 more
Caused by: org.apache.hadoop.hbase.exceptions.DeserializationException: parseFrom called on base Filter, but should be called on derived type
at org.apache.hadoop.hbase.filter.Filter.parseFrom(Filter.java:270)
... 13 more

at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:422)
at org.apache.hadoop.ipc.RemoteException.instantiateException(RemoteException.java:106)
at org.apache.hadoop.ipc.RemoteException.unwrapRemoteException(RemoteException.java:95)
at org.apache.hadoop.hbase.protobuf.ProtobufUtil.getRemoteException(ProtobufUtil.java:329)
at org.apache.hadoop.hbase.client.ScannerCallable.openScanner(ScannerCallable.java:408)
at org.apache.hadoop.hbase.client.ScannerCallable.call(ScannerCallable.java:204)
at org.apache.hadoop.hbase.client.ScannerCallable.call(ScannerCallable.java:65)
at org.apache.hadoop.hbase.client.RpcRetryingCaller.callWithoutRetries(RpcRetryingCaller.java:210)
at org.apache.hadoop.hbase.client.ScannerCallableWithReplicas$RetryingRPC.call(ScannerCallableWithReplicas.java:364)
at org.apache.hadoop.hbase.client.ScannerCallableWithReplicas$RetryingRPC.call(ScannerCallableWithReplicas.java:338)
at org.apache.hadoop.hbase.client.RpcRetryingCaller.callWithRetries(RpcRetryingCaller.java:136)
at org.apache.hadoop.hbase.client.ResultBoundedCompletionService$QueueingFuture.run(ResultBoundedCompletionService.java:65)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Caused by: org.apache.hadoop.hbase.ipc.RemoteWithExtrasException(org.apache.hadoop.hbase.DoNotRetryIOException): org.apache.hadoop.hbase.DoNotRetryIOException: java.lang.reflect.InvocationTargetException
at org.apache.hadoop.hbase.protobuf.ProtobufUtil.toFilter(ProtobufUtil.java:1478)
at org.apache.hadoop.hbase.protobuf.ProtobufUtil.toScan(ProtobufUtil.java:993)
at org.apache.hadoop.hbase.regionserver.RSRpcServices.scan(RSRpcServices.java:2396)
at org.apache.hadoop.hbase.protobuf.generated.ClientProtos$ClientService$2.callBlockingMethod(ClientProtos.java:33648)
at org.apache.hadoop.hbase.ipc.RpcServer.call(RpcServer.java:2180)
at org.apache.hadoop.hbase.ipc.CallRunner.run(CallRunner.java:112)
at org.apache.hadoop.hbase.ipc.RpcExecutor.consumerLoop(RpcExecutor.java:133)
at org.apache.hadoop.hbase.ipc.RpcExecutor$1.run(RpcExecutor.java:108)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.reflect.InvocationTargetException
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.hadoop.hbase.protobuf.ProtobufUtil.toFilter(ProtobufUtil.java:1474)
... 8 more
Caused by: org.apache.hadoop.hbase.exceptions.DeserializationException: parseFrom called on base Filter, but should be called on derived type
at org.apache.hadoop.hbase.filter.Filter.parseFrom(Filter.java:270)
... 13 more

at org.apache.hadoop.hbase.ipc.RpcClientImpl.call(RpcClientImpl.java:1267)
at org.apache.hadoop.hbase.ipc.AbstractRpcClient.callBlockingMethod(AbstractRpcClient.java:227)
at org.apache.hadoop.hbase.ipc.AbstractRpcClient$BlockingRpcChannelImplementation.callBlockingMethod(AbstractRpcClient.java:336)
at org.apache.hadoop.hbase.protobuf.generated.ClientProtos$ClientService$BlockingStub.scan(ClientProtos.java:34094)
at org.apache.hadoop.hbase.client.ScannerCallable.openScanner(ScannerCallable.java:400)
... 10 more

有谁能帮我。。。

Javahadoophbasehbase-filter

2条答案

按热度按时间
1sbrub3j

1sbrub3j1#

只是为了补充一个（也许是重要的）细节，从阿披实库马尔以上的答案。似乎是序列化程序 toByteArray() 和反序列化程序 parseFrom(byte[] rawBytes) 必须通过google协议缓冲区2实现。下面是一个示例实现。
过滤器.java

// whatever fields you need for the AFilter
long fieldA;
long fieldB;   

/**

* Transform this {@code AFilter} instance to a byte array for serialization.
* @return raw bytes of this instance
* /

@Override
public byte[] toByteArray() {
    final FilterProtos.AFilter.Builder builder = FilterProtos.AFilter.newBuilder();
    builder.setFieldA(fieldA);
    builder.setFieldB(fieldB);

    return builder.build().toByteArray();
}

/**
 * De-serialize {@code AFilter} from {@code rawBytes}.
 *
 * @param rawBytes raw bytes of the filter
 * @return AFilter object
 * @throws DeserializationException
 */
public static AFilter parseFrom(final byte[] rawBytes)
        throws DeserializationException {

    try {
        FilterProtos.AFilter proto;
        proto = FilterProtos.AFilter.parseFrom(rawBytes);

        return new AFilter(proto.getFieldA(), proto.getFieldB());
    } catch (InvalidProtocolBufferException ex) {
        throw new DeserializationException(
                ex);
    }
}

过滤器.proto

option java_package = "my.java.package";
option java_outer_classname = "FilterProtos";
option java_generic_services = true;
option java_generate_equals_and_hash = true;
option optimize_for = SPEED;

message AFilter{
    required uint64 fieldA = 1;
    required uint64 fieldB = 2;
}
赞(0）分享  回复(0）举报  2021-05-29
aydmsdu9

aydmsdu92#

正如您在异常堆栈中看到的，它是由于 parseFrom 在基筛选器上调用，但应在派生类型上调用”。
这意味着你必须实施 parseFrom 在自定义过滤器类中也是如此。
另外，您可能需要实现 toByteArray 以及它们一起使用。

赞(0）分享  回复(0）举报  2021-05-29
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com