Apache Cassandra和Apache Ignite:通过Ignite增强Apache Cassandra

服务器

浏览数:61

2020-6-13

Apache Cassandra是开源分布式NoSQL磁盘数据库的领导者之一,作为关键的基础设施,已经部署在诸如Netflix、eBay、Expedia等很多公司中,它因为速度、可线性扩展至上千个节点、一流的数据中心复制而广受欢迎。

Apache Ignite是一个以内存为中心的分布式数据库、缓存和处理平台,可以针对PB级的数据,以内存级的速度处理事务、分析和流式负载,支持JCache、SQL99、ACID事务以及机器学习。

Apache Cassandra在它的领域,是一个经典的解决方案,和任何特定领域解决方案一样,它的优势是建立在一些妥协之上的,一个典型的因素就是受到磁盘存储的限制,Cassandra已经做了尽可能多的优化来解决这些问题。

举个权衡的例子:缺乏ACID和SQL支持之后,就无法随意进行事务和分析,如果数据事先没有进行很好的适配,这些妥协因素,就会对用户造成逻辑上的困扰,导致产品的不正确使用,甚至负体验,或者导致数据在不同类型的存储之间共享,基础设施碎片化以及应用的数据逻辑复杂化。

但是,作为Cassandra的用户,是否可以将其与Apache Ignite一起使用呢?作为前提,目的是维护既有的Cassandra系统然后解决它的局限性,答案是:是,我们可以将Ignite作为一个内存层,部署在Cassandra之上,本文之后就会介绍如何实现。

1.Cassandra的限制

首先,先简要地过一下主要的限制,这些是我们要解决的问题:

  1. 受到磁盘或者SSD特性的限制,带宽和响应时间受到限制;
  2. 数据结构为顺序地读和写进行了优化,没有为传统关系型数据操作的性能优化进行适配,他无法进行数据的标准化以及高效地进行关联操作,并对诸如GROUP BY以及ORDER等进行了严格的限制;
  3. 因为第二条的原因,缺乏对SQL的支持也导致CQL的功能很有限;
  4. 缺少ACID事务;

虽然可以将Cassandra用于其他的用途,那也没问题,但是如果能解决这些问题,会显著增强Cassandra的能力。通过组合人和马,可以得到一个骑手,这相对于单独的人和马,已经是一个完全不同的事物。

那么如何规避这些限制呢?

传统的方法是分割数据,一部分存储于Cassandra中,其他的Cassandra无法保证的部分,存储于不同的系统中。

这个方法的缺点是增加了复杂性(潜在地也可能导致速度以及质量的下降)和增加了维护成本,和使用一个系统做数据存储相比,应用需要组合处理来自不同数据源的数据,甚至,任何其他系统的弱化都可能产生严重的负面影响,迫使基础设施团队疲于奔命。

2.Apache Ignite作为一个内存层

另一个方法是将另一个系统放到Cassandra之上,责任划分之后,Ignite可以提供如下的能力:

  1. 解决了由于磁盘带来的性能限制:Ignite完全在内存中运行,这是目前最快和最廉价的存储之一!
  2. 完全支持标准的SQL99:包括关联、GROUP BY、ORDER BY以及DML,可以标准化数据,便于分析,关注内存性能,打开HTAP的潜力,对生产数据进行实时分析;
  3. 支持JDBC和ODBC标准:便于和已有工具集成,比如Tableau,以及像Hibernate或者Spring Data这样的框架;
  4. 支持ACID事务:如果必须要求一致性,那么这个就是必要的;
  5. 分布式计算、流式数据处理、机器学习:可以使用Ignite提供的技术红利快速实现很多新的业务场景。

Ignite集群使用Cassandra的数据进行查询,开启通写之后会将所有的数据变更回写到Cassandra,下一步,Ignite中持有了数据,就可以自由地使用SQL、运行事务以及享受内存级的速度。

此外,数据也可以使用比如Tableau这样的可视化工具进行实时的分析。

3.配置

下一步,通过一个Ignite和Cassandra集成的简单示例,来说明它们如何一起工作以及可以获得那些特性。

首先,在Cassandra中创建必要的表,注入一些数据,然后初始化一个Java工程,写一些DTO类,然后就会展示核心部分:配置Ignite与Cassandra一起工作。

示例采用了macOS,Cassandra3.10以及Ignite2.3,在Linux中,命令类似。

3.1.Cassandra的表和数据

首先,将Cassandra的发行包放在~ / Downloads文件夹,然后进入该文件夹解压:

$ cd ~/Downloads
$ tar xzvf apache-cassandra-3.10-bin.tar.gz
$ cd apache-cassandra-3.10

使用默认配置启动Cassandra,用于测试这样够了:

$ bin/cassandra

下一步,使用Cassandra的交互式终端创建测试用的数据结构。这里使用常见的id作为主键,对于Cassandra中的表,主键的选择很重要,它关系到后续的数据提取,但是本示例中做了简化:

$ cd ~/Downloads/apache-cassandra-3.10
$ bin/cqlsh
CREATE KEYSPACE IgniteTest WITH replication = {'class': 'SimpleStrategy', 'replication_factor' : 1};

USE IgniteTest;

CREATE TABLE catalog_category (id bigint primary key, parent_id bigint, name text, description text);
CREATE TABLE catalog_good (id bigint primary key, categoryId bigint, name text, description text, price bigint, oldPrice bigint);

INSERT INTO catalog_category (id, parentId, name, description) VALUES (1, NULL, 'Appliances', 'Appliances for households!');
INSERT INTO catalog_category (id, parentId, name, description) VALUES (2, 1, 'Refrigirators', 'The best fridges we have!');
INSERT INTO catalog_category (id, parentId, name, description) VALUES (3, 1, 'Washing machines', 'Engineered for exceptional usage!');

INSERT INTO catalog_good (id, categoryId, name, description, price, oldPrice) VALUES (1, 2, 'Fridge Buzzword', 'Best fridge of 2027!', 1000, NULL);
INSERT INTO catalog_good (id, categoryId, name, description, price, oldPrice) VALUES (2, 2, 'Fridge Foobar', 'The cheapest offer!', 300, 900);
INSERT INTO catalog_good (id, categoryId, name, description, price, oldPrice) VALUES (3, 2, 'Fridge Barbaz', 'Premium fridge in your kitchen!', 500000, 300000);
INSERT INTO catalog_good (id, categoryId, name, description, price, oldPrice) VALUES (4, 3, 'Appliance Habr#', 'Washes, squeezes, dries!', 10000, NULL);

检查一下保存的数据是否正确:

cqlsh:ignitetest> SELECT * FROM catalog_category;

id | description | name | parentId
----+--------------------------------------------+--------------------+-----------
1 | Appliances for households! | Appliances | null
2 | The best fridges we have! | Refrigirators | 1
3 | Engineered for exceptional usage! | Washing machines | 1

(3 rows)
cqlsh:ignitetest> SELECT * FROM catalog_good;

id | categoryId | description | name | oldPrice | price
----+-------------+---------------------------+----------------------+-----------+--------
1 | 2 | Best fridge of 2027! | Fridge Buzzword | null | 1000
2 | 2 | The cheapest offer! | Fridge Foobar | 900 | 300
4 | 3 | Washes, squeezes, dries! | Appliance Habr# | null | 10000
3 | 2 | Premium fridge in your kitchen! | Fridge Barbaz | 300000 | 500000

(4 rows)

3.2.初始化Java工程

Ignite的使用有两种方式:一个是从官网下载发行包,然后将jar文件加入类路径并且使用XML进行配置,或者将Ignite作为Java工程的Maven依赖,本文会使用第二种方式。

使用Maven创建一个新的工程然后加入如下的库:

  • ignite-cassandra-store:用于Cassandra集成;
  • ignite-spring:使用Spring XML文件配置Ignite。

这两个库都依赖ignite-core,它包含了Ignite的核心功能:

<dependencies>
    <dependency>
        <groupId>org.apache.ignite</groupId>
        <artifactId>ignite-spring</artifactId>
        <version>2.3.0</version>
    </dependency>

    <dependency>
        <groupId>org.apache.ignite</groupId>
        <artifactId>ignite-cassandra-store</artifactId>
        <version>2.3.0</version>
    </dependency>
</dependencies>

下一步,创建DTO类,用于映射Cassandra的表:

import org.apache.ignite.cache.query.annotations.QuerySqlField;

public class CatalogCategory {
    @QuerySqlField private long id;
    @QuerySqlField private Long parentId;
    @QuerySqlField private String name;
    @QuerySqlField private String description;

    // public getters and setters
}

public class CatalogGood {
    @QuerySqlField private long id;
    @QuerySqlField private long categoryId;
    @QuerySqlField private String name;
    @QuerySqlField private String description;
    @QuerySqlField private long price;
    @QuerySqlField private long oldPrice;

    // public getters and setters
}

在这些属性上添加了@QuerySqlField注解是为了可以通过Ignite SQL查询到,如果一个属性未加注该注解,它是无法通过SQL进行提取或者通过它进行过滤的。

当然还可以进行微调,比如定义索引以及全文检索,但是这超出了本文的范围,配置Ignite SQL的更多信息,可以查看对应的文档

3.3.配置Ignite

src/main/resources目录中创建一个名为apacheignite-cassandra.xml的配置文件,下面是完整的配置,关键部分后续还会说明:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">

    <bean class="org.apache.ignite.cache.store.cassandra.datasource.DataSource" name="cassandra">
        <property name="contactPoints" value="127.0.0.1"/>
    </bean>

    <bean class="org.apache.ignite.configuration.IgniteConfiguration">
        <property name="cacheConfiguration">
            <list>
                <bean class="org.apache.ignite.configuration.CacheConfiguration">
                    <property name="name" value="CatalogCategory"/>
                    <property name="writeThrough" value="true"/>
                    <property name="sqlSchema" value="catalog_category"/>
                    <property name="indexedTypes">
                        <list>
                            <value type="java.lang.Class">java.lang.Long</value>
                            <value type="java.lang.Class">com.gridgain.test.model.CatalogCategory</value>
                        </list>
                    </property>
                    <property name="cacheStoreFactory">
                        <bean class="org.apache.ignite.cache.store.cassandra.CassandraCacheStoreFactory">
                            <property name="dataSource" ref="cassandra"/>
                            <property name="persistenceSettings">
                                <bean class="org.apache.ignite.cache.store.cassandra.persistence.KeyValuePersistenceSettings">
                                    <constructor-arg type="java.lang.String"><value><![CDATA[
                                        <persistence keyspace="IgniteTest" table="catalog_category">
                                            <keyPersistence class="java.lang.Long" strategy="PRIMITIVE" column="id"/>
                                            <valuePersistence class="com.gridgain.test.model.CatalogCategory" strategy="POJO"/>
                                        </persistence>]]></value></constructor-arg>
                                </bean>
                            </property>
                        </bean>
                    </property>
                </bean>

                <bean class="org.apache.ignite.configuration.CacheConfiguration">
                    <property name="name" value="CatalogGood"/>
                    <property name="readThrough" value="true"/>
                    <property name="writeThrough" value="true"/>
                    <property name="sqlSchema" value="catalog_good"/>
                    <property name="indexedTypes">
                        <list>
                            <value type="java.lang.Class">java.lang.Long</value>
                            <value type="java.lang.Class">com.gridgain.test.model.CatalogGood</value>
                        </list>
                    </property>
                    <property name="cacheStoreFactory">
                        <bean class="org.apache.ignite.cache.store.cassandra.CassandraCacheStoreFactory">
                            <property name="dataSource" ref="cassandra"/>
                            <property name="persistenceSettings">
                                <bean class="org.apache.ignite.cache.store.cassandra.persistence.KeyValuePersistenceSettings">
                                    <constructor-arg type="java.lang.String"><value><![CDATA[
                                        <persistence keyspace="IgniteTest" table="catalog_good">
                                            <keyPersistence class="java.lang.Long" strategy="PRIMITIVE" column="id"/>
                                            <valuePersistence class="com.gridgain.test.model.CatalogGood" strategy="POJO"/>
                                        </persistence>]]></value></constructor-arg>
                                </bean>
                            </property>
                        </bean>
                    </property>
                </bean>
            </list>
        </property>
    </bean>

</beans>

上述配置可以分为两个部分,首先,定义一个连接Cassandra的数据源,第二是Ignite本身的配置。 第一部分的配置比较简单:

 <bean class="org.apache.ignite.cache.store.cassandra.datasource.DataSource" name="cassandra">
        <property name="contactPoints" value="127.0.0.1"/>
    </bean>

使用IP地址来定义要连接的Cassandra数据源。

下一步要配置Ignite,在本例中,和默认的配置相比只有很小的区别,只是覆写了cacheConfiguration属性,它包含了一组映射到Cassandra表的Ignite缓存:

<bean class="org.apache.ignite.configuration.IgniteConfiguration">
        <property name="cacheConfiguration">
            <list>
                ...
            </list>
        </property>
    </bean>

第一个缓存映射到Cassandra的catalog_category表:

<bean class="org.apache.ignite.configuration.CacheConfiguration">
	<property name="name" value="CatalogCategory"/>
	...
</bean>

每个缓存都开启了通读和通写模式,比如,如果对Ignite执行了写入操作,那么Ignite会自动发送一个更新操作给Cassandra,接下来,指定了在Ignite中使用catalog_category模式:

<property name="readThrough" value="true"/>
<property name="writeThrough" value="true"/>
<property name="sqlSchema" value="catalog_category"/>
<property name="indexedTypes">
	<list>
		<value type="java.lang.Class">java.lang.Long</value>
		<value type="java.lang.Class">com.gridgain.test.model.CatalogCategory</value>
	</list>
</property>

最后,建立到Cassandra的连接,这里面有两个主要的子片段,首先,要指向之前创建的数据源,然后,要将Ignite缓存和Cassandra的表建立关联。

本来,通过persistenceSettings属性指向一个外部的配置映射的XML配置文件是比较好的,但是为了简化,将这段XML作为CDATA片段直接嵌入了Spring的配置文件:

<property name="cacheStoreFactory">
	<bean class="org.apache.ignite.cache.store.cassandra.CassandraCacheStoreFactory">
		<property name="dataSource" ref="cassandra"/>
		<property name="persistenceSettings">
			<bean class="org.apache.ignite.cache.store.cassandra.persistence.KeyValuePersistenceSettings">
				<constructor-arg type="java.lang.String"><value><![CDATA[
					<persistence keyspace="IgniteTest" table="catalog_category">
						<keyPersistence class="java.lang.Long" strategy="PRIMITIVE" column="id"/>
						<valuePersistence class="com.gridgain.test.model.CatalogCategory" strategy="POJO"/>
					</persistence>]]></value></constructor-arg>
			</bean>
		</property>
	</bean>
</property>

映射的配置看上去非常简单明了:

<persistence keyspace="IgniteTest" table="catalog_category">
    <keyPersistence class="java.lang.Long" strategy="PRIMITIVE" column="id"/>
    <valuePersistence class="com.gridgain.test.model.CatalogCategory" strategy="POJO"/>
</persistence>

在最上层(persistence标签),声明了键空间(本例中为IgniteTest)和要关联的表(catalog_category),然后声明了Ignite缓存的主键为Long类型,这是个基本类型,它对应了Cassandra表中的id列。在本例中,值为CatalogCategory类,借助于反射(策略为POJO),建立了和Cassandra表中的列的关联。

关于映射的更多细节,超出了本文的细节,具体可以看相关的文档

第二部分与产品数据有关的缓存配置,大体相同。

3.4.启动

使用下面的类可以启动:

package com.gridgain.test;

import org.apache.ignite.Ignite;
import org.apache.ignite.Ignition;

public class Starter {
    public static void main(String... args) throws Exception {
        final Ignite ignite = Ignition.start("apacheignite-cassandra.xml");

        ignite.cache("CatalogCategory").loadCache(null);
        ignite.cache("CatalogGood").loadCache(null);
    }
}

这里使用了Ignition.start(...)方法来启动一个Ignite节点,ignite.cache(...).loadCache(null)方法用于将Cassandra中的数据预加载到Ignite中。

3.5.SQL

Ignite集群启动,接入Cassandra之后,就可以执行Ignite SQL查询了。比如,可以使用任何支持JDBC或者ODBC的客户端,在本例中使用了SquirrelSQL,首先需要为工具添加Ignite的JDBC驱动 使用jdbc:ignite://localhost/CatalogGood这样的URL形式建立一个连接,这里localhost是Ignite集群中一个节点的地址,然后CatalogGood是默认请求的缓存名。 最后,可以执行几个SQL查询:

SELECT cg.name goodName, cg.price goodPrice, cc.name category, pcc.name parentCategory
FROM catalog_category.CatalogCategory cc
  JOIN catalog_category.CatalogCategory pcc
  ON cc.parentId = pcc.id
  JOIN catalog_good.CatalogGood cg
  ON cg.categoryId = cc.id;
goodName goodPrice category parentCategory
Fridge Buzzword 1000 Refrigerators Appliances
Fridge Foobar 300 Refrigerators Appliances
Fridge Barbaz 500,000 Refrigerators Appliances
Appliance Habr # 10000 Washing machines Appliances
SELECT cc.name, AVG(cg.price) avgPrice
FROM catalog_category.CatalogCategory cc
  JOIN catalog_good.CatalogGood cg
  ON cg.categoryId = cc.id
WHERE cg.price <= 100000
GROUP BY cc.id;
name avgPrice
Refrigerators 650
Washing machines 10000

4.结论

在这个简单的示例中,展示了如何通过引入Ignite,为已有的Cassandra系统带来了内存级性能的SQL功能。因此,如果正受到本文提到的Cassandra限制的困扰,那么就需要考虑一下Ignite这个备选的技术方案。

本文译自GridGain的业务架构师Artem Schitow的博客

作者:李玉珏