数据税Cassandra散货船

Question

我正在尝试使用JMX BulkLoader将ETL数据从远程节点传输到集群上的Cassandra

https://github.com/PatrickCallaghan/datastax-analytics-example/blob/master/src/main/java/com/datastax/jmxloader/JmxBulkLoader.java

但是，在成功建立JMX连接后，它似乎无法批量加载。

请注意，批量负载是从远程节点发布到cassandra集群的。

它看起来似乎期望在cassandra集群的本地运行(即cassandra集群的本地主机)

我错过了什么吗？有人能给点建议吗

以下例外情况

java.lang.IllegalArgumentException:位于org.apache.casandra.service.StorageService.bulkLoadInternal（StorageService.java:3970）位于org.apache/casandra.server.StorageServices.bulkLoad Async（StorageService.java:3962）位于sun.relect.GeneratedMethodAccessor21.invoke（未知源）位于sun/relect.DelegatingMethodAccessorImpl.invokesun.reflect.Method.invoke（Method.java：606）at sun.reflekt.misc.Trampoline.invouse（MethodUtil.java：75）at sun/reflect.GeneratedMethodAccessor2.invoke（未知源）at sun-reflect.DelegatingMethodAccessorImpl.invokecom.sun.jmx.mbeanserver.StandardMBeanIntrospector.invokeM2（StandardMBeanIntrosopector.java:112），位于com.sun.jmx.mbeanserver.MBeanSupport.invoke，位于com.sun.jmx.mmbeanserver.PerInterface.invoke（PerInterface.java:138）（MBeanSupport.java：252）位于com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.invoke（DefaultMBeanServerInterceptor.java：819）位于com.sun.jmx.mmbeanserver.JmxMBeanServer.invoke。

class JmxBulkLoader(host: String, port: Int) {

  private var connector: JMXConnector = _

  private var storageBean: StorageServiceMBean = _

  private var timer: Timer = new Timer()

  connect("http://hostip , 7199)

 private def connect(host: String, port: Int) {

    val jmxUrl = new JMXServiceURL("service:jmx:rmi:///jndi/rmi://%s:%d/jmxrmi".format(host,

      port))

    Logger.info(" Connected to JMX Entity " + jmxUrl)

    val env = new HashMap[String, Any]()

    connector = JMXConnectorFactory.connect(jmxUrl, env)

    val mbeanServerConn = connector.getMBeanServerConnection

    val name = new ObjectName("org.apache.cassandra.db:type=StorageService")

    storageBean = JMX.newMBeanProxy(mbeanServerConn, name, classOf[StorageServiceMBean])

  }

  def close() {

    connector.close()

  }

  def bulkLoad(path: String): Boolean = {

    try {

      val timer = new Stopwatch().start

      val result = storageBean.bulkLoadAsync(path)

      timer.stop

      Logger.info("Async Result of Bulk Load " + result)

      Logger.info("Bulk load took " + timer.getElapsedTime + "millsecs.")

      true

    } catch {

      case e: Exception =>

        Logger.error("Error in Bulk Loading " + e.printStackTrace())

        false

    }

  }

}

Answer 1

1. 该表应存在于卡桑德拉中
2. 该目录对于 cassandra 节点应该是可访问的（本地）。
3. 目录应以键空间和目标表名结尾：
   /some_path/$KeySpaceName/$TableName

Answer 2

它看起来似乎期望在cassandra集群的本地运行(即cassandra集群的本地主机)

差一点。但是想想看：你用一个字符串参数调用Cassandra节点的mbean函数。此调用由您正在调用的 Cassandra 进程执行（即连接到）。该参数指定要连接到的节点一侧的路径。

您必须确保路径存在于目标上并保存您期望的数据（例如，通过共享存储或事先复制文件）。