使用自定义过滤器过滤 Accumulo 返回的结果时出错

【问题标题】：Error when using a custom filter to filter results returned by Accumulo使用自定义过滤器过滤 Accumulo 返回的结果时出错
【发布时间】：2016-02-11 08:40:27
【问题描述】：

我编写了一个非常简单的自定义过滤器来过滤 Accumulo 返回的结果。这是我写的过滤器

public class MyFilter extends Filter {

    @Override
    public boolean accept(Key key, Value val) {
        Long page = 1L;
        Integer limit = 25;

        if(key.getColumnQualifier().getBytes().equals("Class".getBytes()) && val.get().equals("1".getBytes())) {
            if(page == 1) {
                return true;
            }
            limit--;
            if(limit == 1L) {
                page++;
                limit = 25;
            }
        }

        return false;
    }
}

我像这样将这个过滤器添加到 Accumulo 的扫描仪中

Set<Range> ranges = new HashSet<>();
IteratorSetting iter = new IteratorSetting(15, "MyFilter", MyFilter.class);
myScanner.addScanIterator(iter);
Iterator<Entry<Key, Value>> kys = myScanner.iterator();
while(kys.hasNext()) { // This is line 335 in com.latize.ulysses3.service.AccumuloPivotTable.getRows
    Entry<Key, Value> e = kys.next();
    ranges.add(Range.exact(e.getKey().getRow()));
}

但是每当我尝试运行这段代码时，我都会得到这个堆栈跟踪

java.lang.RuntimeException: org.apache.accumulo.core.client.impl.AccumuloServerException: Error on server accumulo.tablet.2:9997
        at org.apache.accumulo.core.client.impl.ScannerIterator.hasNext(ScannerIterator.java:161)
        at com.latize.ulysses3.service.AccumuloPivotTable.getRows(AccumuloPivotTable.java:335)
        at com.latize.ulysses3.webservice.DataVault.getFilteredRows(DataVault.java:950)
        at com.latize.ulysses3.webservice.DataVault$Proxy$_$$_WeldClientProxy.getFilteredRows(Unknown Source)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:606)
        at org.jboss.resteasy.core.MethodInjectorImpl.invoke(MethodInjectorImpl.java:137)
        at org.jboss.resteasy.core.ResourceMethodInvoker.invokeOnTarget(ResourceMethodInvoker.java:280)
        at org.jboss.resteasy.core.ResourceMethodInvoker.invoke(ResourceMethodInvoker.java:234)
        at org.jboss.resteasy.core.ResourceMethodInvoker.invoke(ResourceMethodInvoker.java:221)
        at org.jboss.resteasy.core.SynchronousDispatcher.invoke(SynchronousDispatcher.java:356)
        at org.jboss.resteasy.core.SynchronousDispatcher.invokePropagateNotFound(SynchronousDispatcher.java:217)
        at org.jboss.resteasy.plugins.server.servlet.ServletContainerDispatcher.service(ServletContainerDispatcher.java:224)
        at org.jboss.resteasy.plugins.server.servlet.FilterDispatcher.doFilter(FilterDispatcher.java:62)
        at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:241)
        at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:208)
        at org.apache.marmotta.platform.core.servlet.MarmottaResourceFilter$LMFFilterChain.doFilter(MarmottaResourceFilter.java:189)
        at org.apache.marmotta.platform.core.filters.ModuleResourceFilter.doFilter(ModuleResourceFilter.java:169)
        at org.apache.marmotta.platform.core.filters.ModuleResourceFilter$Proxy$_$$_WeldClientProxy.doFilter(Unknown Source)
        at org.apache.marmotta.platform.core.servlet.MarmottaResourceFilter$LMFFilterChain.doFilter(MarmottaResourceFilter.java:184)
        at com.latize.ulysses.platform.core.filters.ClickJackFilter.doFilter(ClickJackFilter.java:105)
        at com.latize.ulysses.platform.core.filters.ClickJackFilter$Proxy$_$$_WeldClientProxy.doFilter(Unknown Source)
        at org.apache.marmotta.platform.core.servlet.MarmottaResourceFilter$LMFFilterChain.doFilter(MarmottaResourceFilter.java:184)
        at com.latize.ulysses.platform.core.filters.SessionTimeoutCookieFilter.doFilter(SessionTimeoutCookieFilter.java:125)
        at com.latize.ulysses.platform.core.filters.SessionTimeoutCookieFilter$Proxy$_$$_WeldClientProxy.doFilter(Unknown Source)
        at org.apache.marmotta.platform.core.servlet.MarmottaResourceFilter$LMFFilterChain.doFilter(MarmottaResourceFilter.java:184)
        at org.apache.marmotta.platform.core.filters.MarmottaServerNameFilter.doFilter(MarmottaServerNameFilter.java:104)
        at org.apache.marmotta.platform.core.servlet.MarmottaResourceFilter$LMFFilterChain.doFilter(MarmottaResourceFilter.java:184)
        at com.latize.ulysses.platform.core.filters.UlyssesTemplatingFilter.doFilter(UlyssesTemplatingFilter.java:178)
        at com.latize.ulysses.platform.core.filters.UlyssesTemplatingFilter$Proxy$_$$_WeldClientProxy.doFilter(Unknown Source)
        at org.apache.marmotta.platform.core.servlet.MarmottaResourceFilter$LMFFilterChain.doFilter(MarmottaResourceFilter.java:184)
        at org.apache.marmotta.platform.core.filters.TemplatingFilter.doFilter(TemplatingFilter.java:176)
        at org.apache.marmotta.platform.core.filters.TemplatingFilter$Proxy$_$$_WeldClientProxy.doFilter(Unknown Source)
        at org.apache.marmotta.platform.core.servlet.MarmottaResourceFilter$LMFFilterChain.doFilter(MarmottaResourceFilter.java:184)
        at org.apache.marmotta.platform.security.filters.MarmottaAccessControlFilter.doFilter(MarmottaAccessControlFilter.java:142)
        at org.apache.marmotta.platform.security.filters.MarmottaAccessControlFilter$Proxy$_$$_WeldClientProxy.doFilter(Unknown Source)
        at org.apache.marmotta.platform.core.servlet.MarmottaResourceFilter$LMFFilterChain.doFilter(MarmottaResourceFilter.java:184)
        at org.apache.marmotta.platform.user.filters.MarmottaAuthenticationFilter.doFilter(MarmottaAuthenticationFilter.java:228)
        at org.apache.marmotta.platform.user.filters.MarmottaAuthenticationFilter$Proxy$_$$_WeldClientProxy.doFilter(Unknown Source)
        at org.apache.marmotta.platform.core.servlet.MarmottaResourceFilter$LMFFilterChain.doFilter(MarmottaResourceFilter.java:184)
        at org.apache.marmotta.platform.core.servlet.MarmottaResourceFilter.doFilter(MarmottaResourceFilter.java:135)
        at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:241)
        at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:208)
        at org.apache.marmotta.platform.core.servlet.MarmottaPreStartupFilter.doFilter(MarmottaPreStartupFilter.java:106)
        at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:241)
        at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:208)
        at com.thetransactioncompany.cors.CORSFilter.doFilter(CORSFilter.java:169)
        at com.thetransactioncompany.cors.CORSFilter.doFilter(CORSFilter.java:232)
        at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:241)
        at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:208)
        at org.apache.marmotta.platform.core.servlet.MarmottaOptionsFilter.doFilter(MarmottaOptionsFilter.java:83)
        at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:241)
        at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:208)
        at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:220)
        at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:122)
        at org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:505)
        at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:170)
        at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:103)
        at org.apache.catalina.valves.AccessLogValve.invoke(AccessLogValve.java:956)
        at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:116)
        at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:423)
        at org.apache.coyote.http11.AbstractHttp11Processor.process(AbstractHttp11Processor.java:1079)
        at org.apache.coyote.AbstractProtocol$AbstractConnectionHandler.process(AbstractProtocol.java:625)
        at org.apache.tomcat.util.net.JIoEndpoint$SocketProcessor.run(JIoEndpoint.java:318)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
        at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61)
        at java.lang.Thread.run(Thread.java:745)
Caused by: org.apache.accumulo.core.client.impl.AccumuloServerException: Error on server accumulo.tablet.2:9997
        at org.apache.accumulo.core.client.impl.ThriftScanner.scan(ThriftScanner.java:293)
        at org.apache.accumulo.core.client.impl.ScannerIterator$Reader.run(ScannerIterator.java:80)
        at org.apache.accumulo.core.client.impl.ScannerIterator.hasNext(ScannerIterator.java:151)
        ... 69 more
Caused by: org.apache.thrift.TApplicationException: Internal error processing startScan
        at org.apache.thrift.TApplicationException.read(TApplicationException.java:111)
        at org.apache.thrift.TServiceClient.receiveBase(TServiceClient.java:71)
        at org.apache.accumulo.core.tabletserver.thrift.TabletClientService$Client.recv_startScan(TabletClientService.java:232)
        at org.apache.accumulo.core.tabletserver.thrift.TabletClientService$Client.startScan(TabletClientService.java:208)
        at org.apache.accumulo.core.client.impl.ThriftScanner.scan(ThriftScanner.java:410)
        at org.apache.accumulo.core.client.impl.ThriftScanner.scan(ThriftScanner.java:285)
        ... 71 more

accumulo.tablet.2:9997 是运行我的一台平板电脑服务器的机器的域名。配置很好，因为其他功能正常。端口也起来了。有人能告诉我为什么会出现这个错误吗？任何帮助将不胜感激。

【问题讨论】：

标签： java hadoop accumulo


【解决方案1】：

要正确使用过滤器，您必须将它们添加到 accumulo 类路径中，通常通过部署 .jar 文件，其中迭代器在系统的每个节点上定义为 $ACCUMULO_HOME/lib 或 $ACCUMULO_HOME/lib/ext。为了更深入地了解迭代器，我推荐非常好的documentation of accumulo。

为了便于部署，我建议使用Jenkins、Puppet 或Chef。如果这对您的项目来说很重要，那么一个简单的 shell 脚本也可以。您可能会发现cluster shell cssh 对管理问题也很有帮助！请注意，在 ext 文件夹中部署后，您确实不需要重新启动 accumulo，但是对于 lib 文件夹，这是必需的。

【讨论】：

  • 如果你的集群上有很多节点，那么你必须手动为每个节点添加过滤器吗？有没有办法将 Accumulo 的类路径设置为 HDFS 位置，以便所有节点都可以访问它，而无需我在每个节点上手动部署它们。
  • 问题是我有一个应用程序可以查询 Accumulo 中的一堆表。此应用程序（基于 Web）使用 Accumulo 的 Java API 来查询集群。应用程序本身不在托管任何平板服务器的任何机器上运行。如果我需要将过滤器显式部署到集群中的每个节点，那么我还必须将其添加到我的应用程序的类路径中。对我来说似乎很麻烦。
  • 我在回答中添加了更多细节。如果您想直接从您的 web 应用程序访问 accumulo，您必须添加依赖项，这是肯定的。如果你必须保持 webapp 的精简，你可能想在你的集群中添加一个简单的 rest 服务器来与 web 应用程序通信。尽管这会增加一些努力......
  • 澄清一下：在$ACCUMULO_HOME/lib 中放置一个罐子确实需要重新启动。 $ACCUMULO_HOME/lib/ext 没有。
  • 非常感谢您的澄清
【解决方案2】：

我有问题。当我检查 tserver 的日志时，我得到了一个 ClassNotFound 异常。看来我需要将该类添加到 Accumulo 的类路径中。但是，如果我想将这些过滤结果作为应用程序的一部分而不是从终端运行它，我将如何去做呢？我可能无法将过滤器添加到 Accumulo 的类路径中。

【讨论】：

    猜你喜欢
    • 1970-01-01
    • 2018-03-15
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2017-08-07
    • 1970-01-01
    • 1970-01-01
    相关资源
    最近更新 更多
    热门标签
    Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode