时间戳

首页 > 解决方案 > Cassandra-ignite 持久性不起作用

cassandra - Cassandra-ignite 持久性不起作用

问题描述

我正在将 ignite 缓存与 Cassandra 集成。所以我用持久性配置创建了一个名为“cache1”的点燃缓存。并在缓存中添加了一些值，当时它正在工作。但是当我尝试从 spark-shell 或 spark-submit 访问 ignite 缓存“cache1”时出现错误，同时访问集群中的其他缓存时没有问题。

我已经做了以下事情，但我仍然收到错误。

  1. 添加了在配置中启用的对等类
  2. 添加了可选库到 ignite/libs/

错误消息是

019-10-08 08:54:23 ERROR TcpDiscoverySpi:586 - Failed to unmarshal discovery data for component: 1
class org.apache.ignite.IgniteCheckedException: Failed to find class with given class loader for unmarshalling (make sure same versions of all classes are available on all nodes or enable peer-class-loading) [clsLdr=sun.misc.Launcher$AppClassLoader@266474c2, cls=org.apache.ignite.cache.store.cassandra.CassandraCacheStoreFactory]

class org.apache.ignite.IgniteCheckedException: Failed to find class with given class loader for unmarshalling (make sure same versions of all classes are available on all nodes or enable peer-class-loading) [clsLdr=sun.misc.Launcher$AppClassLoader@5e2de80c, cls=org.apache.ignite.cache.store.cassandra.CassandraCacheStoreFactory]
at org.apache.ignite.marshaller.jdk.JdkMarshaller.unmarshal0(JdkMarshaller.java:144)
at org.apache.ignite.marshaller.AbstractNodeNameAwareMarshaller.unmarshal(AbstractNodeNameAwareMarshaller.java:94)
at org.apache.ignite.marshaller.jdk.JdkMarshaller.unmarshal0(JdkMarshaller.java:161)
at org.apache.ignite.marshaller.AbstractNodeNameAwareMarshaller.unmarshal(AbstractNodeNameAwareMarshaller.java:82)
at org.apache.ignite.spi.discovery.tcp.internal.DiscoveryDataPacket.unmarshalData(DiscoveryDataPacket.java:280)
at org.apache.ignite.spi.discovery.tcp.internal.DiscoveryDataPacket.unmarshalGridData(DiscoveryDataPacket.java:123)
at org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi.onExchange(TcpDiscoverySpi.java:2006)
at org.apache.ignite.spi.discovery.tcp.ClientImpl$MessageWorker.processNodeAddFinishedMessage(ClientImpl.java:2181)
at org.apache.ignite.spi.discovery.tcp.ClientImpl$MessageWorker.processDiscoveryMessage(ClientImpl.java:2060)
at org.apache.ignite.spi.discovery.tcp.ClientImpl$MessageWorker.body(ClientImpl.java:1905)
at org.apache.ignite.internal.util.worker.GridWorker.run(GridWorker.java:120)
at org.apache.ignite.spi.discovery.tcp.ClientImpl$1.body(ClientImpl.java:304)
at org.apache.ignite.spi.IgniteSpiThread.run(IgniteSpiThread.java:62)
Caused by: java.lang.ClassNotFoundException: org.apache.ignite.cache.store.cassandra.CassandraCacheStoreFactory
at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:349)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
at java.lang.Class.forName0(Native Method)
at java.lang.Class.forName(Class.java:348)
at org.apache.ignite.internal.util.IgniteUtils.forName(IgniteUtils.java:8771)
at org.apache.ignite.marshaller.jdk.JdkMarshallerObjectInputStream.resolveClass(JdkMarshallerObjectInputStream.java:59)
at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1868)
at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1751)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2042)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1573)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2287)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2211)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2069)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1573)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2287)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2211)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2069)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1573)
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:431)
at java.util.HashMap.readObject(HashMap.java:1412)
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:498)
at java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1170)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2178)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2069)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1573)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2287)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2211)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2069)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1573)
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:431)
at org.apache.ignite.marshaller.jdk.JdkMarshaller.unmarshal0(JdkMarshaller.java:139)
... 12 more

我的持久性设置文件是

<persistence keyspace="test" table="test_table" ttl="86400">
<keyspaceOptions>
    REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor' : 3}
    AND DURABLE_WRITES = true
</keyspaceOptions>
<tableOption>
    comment = 'Cache test'
    AND read_repair_chance = 0.2
</tableOption>
<keyPersistence class="java.lang.String" strategy="PRIMITIVE" column="key"/>
<valuePersistence class="com.test.ignite.Test" strategy="POJO"/>

使用下面的代码启动“cache1”

 public static void main(String[] args) throws IOException {

    //  Handles the configuration configuration

    IgniteConfiguration cfg = new IgniteConfiguration();
    CacheConfiguration configuration = new CacheConfiguration();

    //  Setting cache name
    configuration.setName("cache1");

    //  Configuring Cassandra's persistence
    DataSource dataSource = new DataSource();
    dataSource.setContactPoints("10.0.1.13");
    RoundRobinPolicy robinPolicy = new RoundRobinPolicy();
    dataSource.setLoadBalancingPolicy(robinPolicy);
    dataSource.setReadConsistency("ONE");
    dataSource.setWriteConsistency("ONE");
    String persistenceSettingsXml = FileUtils.readFileToString(new File("/etc/ignite/config/cassandra_config/persistence-settings4.xml"), "utf-8");
    KeyValuePersistenceSettings persistenceSettings = new KeyValuePersistenceSettings(persistenceSettingsXml);
    CassandraCacheStoreFactory cacheStoreFactory = new CassandraCacheStoreFactory();
    cacheStoreFactory.setDataSource(dataSource);
    cacheStoreFactory.setPersistenceSettings(persistenceSettings);
    configuration.setCacheStoreFactory(cacheStoreFactory);
    configuration.setWriteThrough(true);
    configuration.setWriteBehindEnabled(true);
    configuration.setReadThrough(true);
    cfg.setPeerClassLoadingEnabled(true);

    //  Sets the cache configuration
    cfg.setCacheConfiguration(configuration);

    //  Starting Ignite
    Ignite ignite = Ignition.start(cfg);
}

当我执行下面的代码时，我得到了错误。

public class Client {

public static void main(String ars[]){

    //  Start Ignite in client mode.
    Ignition.setClientMode(true);
    Ignite ignite = Ignition.start();

    //  Getting cache
    final IgniteCache<String, Test> cache = ignite.getOrCreateCache("cache1");

    //  Put some data
    cache.put("1", new Test(10,10));

    //  Get some data
    Test val1 = cache.get("1");
    System.out.println("Value: "+ val1);
}
}

标签： cassandraignite

解决方案

此错误与调用 spark-submit 有关。在类路径中添加 ignite-cassandra-store 后它起作用了。现在它正在工作。

推荐阅读