miki艾比利的博客

愿居于一城,与卿所见美好......

您现在的位置是:首页>漫生活

    用Kettle在数据集成中进行分区

    作者:miki艾比利分类:[ETL]浏览(1193评论(0

    让我们假设,我们用简单的数据集进行了一次转换,该数据集具有印度城市明智的州人口数。我们需要获取总人口数。我们使用“ 内存按步骤分组” 构建一个简单的转换文件,并根据State对源数据进行分组以获得总人口数。

    KTR找到明智的人口计数国家

    现在,假设你有少量的源数据,而不是源中的数百万条记录。因此,逐步进行内存分组将花费你巨大的成本。我们不是简单地通过,而是利用多核计算机的功能。

    Pentaho将为你提供一个定义步骤中数据副本数的选项。在Pentaho中按步骤定义内存组的多行是使用多核计算机的一种方法。因此,请按以下步骤对“内存组”进行以下操作:

    >>  右键单击 “内存分组依据
     ” >>选择“更改份数以开始
     ” >>将值设置为 3或大于1的任何数字

    将份数输入到3

    但是,一旦执行了上面的转换,我们就会看到结果有很大的不同,这并不是预期的。这是因为源步骤是按循环方式将数据逐步发送到“内存组”的3个副本。这些内存组中的每一个都以自己的方式逐步处理数据,最后发送输出。这是PDI中的并行性,但是在这种情况下,我们没有得到正确的输出。

    因此,我们在这里要做的就是对数据进行分区。这里的想法是将类似种类的数据发送到每个分区,然后通过分组而不是根据上述数据进行分组。它像3 Memory Step by Step,每个分区都有分区数据,例如分区1中的西孟加拉邦,分区2中的喀拉拉邦,等等。

    在PDI中创建分区

    创建分区要遵循的步骤:

    >>转到转换的
    设计”选项卡>>在“分区模式”选项卡中,创建一个“新模式
     ” >>输入“分区名称”。它可以是任何可读的名称。
    >>输入分区ID或所需的分区数。
    到目前为止,对分区没有命名约束。 
    只需给出分区1,分区2等。

    完成上述步骤后。现在继续进行转换并执行以下操作:

    >>右键单击“内存组”,一步步
     >>选择“分区”。
    >>选择“分区”方法。选择除法余数>>选择分区架构。
    这是上面在“设计”选项卡中创建的所有分区架构的列表。
    >>接下来,你需要选择要分区的字段。
    选择一个合适的。

    最后,你将看到类似以下的内容:

    带有分区架构和数据的KTR

    执行后,返回的结果是正是我们需要的:

    最终分区输出

    全文
2019-11-14 09:38:16

    【转载】kettle集群原理分析

    作者:miki艾比利分类:[ETL]浏览(636评论(0

    本文转自:http://blog.csdn.net/dqswuyundong/article/details/5952009

    Kettle集群

    Kettle是一款开源的ETL工具,以其高效和可扩展性而闻名于业内。其高效的一个重要原因就是其多线程和集群功能。

    Kettle的多线程采用的是一种流水线并发的机制,我们在另外的文章中专门有介绍。这里主要介绍的是kettle的集群。

    集群允许转换以及转换中的步骤在多个服务器上并发执行。在使用kettle集群时,首先需要定义的是Cluster schema。所谓的Cluster schema就是一系列的子服务器的集合。在一个集群中,它包含一个主服务器(Master)和多个从属服务器服务器(slave)。如下图所示。

    子服务器(Slave servers)允许你在远程服务器上执行转换。建立一个子服务器需要你在远程服务器上建立一个叫做“Carte”的 web 服务器,该服务器可以从Spoon(远程或者集群执行)或者转换任务中接受输入。

    在以后的描述中,如果我们提到的是子服务器,则包括集群中的主服务器和从属服务器;否则我们会以主服务器和从属服务器来进行特别指定。

    设计

    要让转换是以集群方式执行,首先需要在Spoon中进行图形化的设计工作。定义一个以集群方式运行的转换,主要包括定义cluster schema和定义转换两个步骤。

    定义cluster schema

    创建子服务器

    服务tab 选项

    选项

    描述

    服务器名称

    子服务器的名称

    主机名称或IP地址

    用作子服务器的机器的地址

    端口号

    与远程服务通信的端口号

    用户名

    获取远程服务器的用户名

    密码

    获取远程服务器的密码

    是主服务器吗

    在转换以集群形式执行时,该子服务器将作为主服务器

    注意: 在集群环境下执行转化时,你必须有一个子服务器作为主服务器(master server)而其余所有的子服务器都作为从属服务器(slave)

    Proxy tab options

    选项

    描述

    代理服务器主机名

    设置你要通过代理进行连接的主机名

    代理服务器端口

    设置与代理进行连接时所需的端口号

       

    Ignore proxy for hosts: regexp|separated

    指定哪些服务器不需要通过代理来进行连接。该选项支持你使用正则表达式来制定多个服务器,多个服务器之间以' | ' 字符来进行分割

    创建cluster schema

    选项描述

    选项

    描述

    Schema 名称

    集群schema的名称

    端口号

    这里定义的端口号是指从哪一个端口号开始分配给子服务器。每一个在子服务器中执行的步骤都要消耗一个端口号。

    注意: 确保没有别的网络协议会使用你定义的范围之类的端口,否则会引起问题

    Sockets缓存大小

    TCP内部缓存的大小

    Sockets刷新间隔(rows)

    当TCP的内部缓存通过网络完全发送出去并且被清空时处理的行数

    Sockets数据是否压缩

    如果该选项被选中,则所有的数据都会使用Gzip压缩算法进行压缩以减轻网络传输量

    Dynamic Cluster

    动态集群指的是在运行的时候才能获知从属服务器的信息。这种情形适用于主机可以自动增加或者去除的情形,例如云计算。

    主服务器的设置不变,但是它可以接受从属服务器的注册。一旦接受了某个从属服务器的注册,则每隔30秒去监视该从属服务器是否还处于有效状态

    子服务器

    这里是一个要在集群中使用的服务器列表。这个列表中包含一个主服务器和任意数目的从属服务器。

    在dynamic Cluster的情况下,只需要选择主服务器即可

    定义转换

    定义完了cluster schema后,下一步就是定义在集群环境下执行的转换。我们这里展现的只是一个最简单的例子,完全是为了演示而用。现实情况中的集群有可能非常复杂。

    首先你像平时一样创建转换,以hop连接连个两个步骤。然后你指定第二个步骤将在集群下执行

    然后选择需要使用的集群。转换如图一样显示在GUI中。

    注意 Cx4显示这个步骤将在集群中运行,而这个集群中有4个从属服务器。假设我们将计算结果再次存入到数据表中

    这个转换虽然定义了集群,但是我们同样可以让它在单机环境下执行,而且可以得到相同的结果。这意味着你可以使用普通的本地模式来测试它。

    执行转换

    要想以集群方式来运行转换或者作业,首先需要启动在Cluster  schema中定义的主服务器和从属服务器,然后再运行转换或者作业。

    启动子服务器

    子服务器其实是一个嵌入式的名为Carte的小web server。要进行集群转换,首先需要启动cluster schema中的子服务器

    脚本启动

    kettle提供了carte.bat和carte.sh(linux)批处理脚本来启动子服务器,这种启动方式分为两种

    使用主机号和端口号

    Carte 127.0.0.1 8080

    Carte 192.168.1.221 8081

    使用配置文件

    Carte  /foo/bar/carte-config.xml

    Carte http://www.example.com/carte-config.xml

    如果cluster schema中定义了Dynamic cluster选项,则必须使用配置文件来进行启动,当这个子服务器启动时,它需要向配置文件中“masters”中列出的主服务器列表中汇报其运行状态(通过调用主服务器的registerSlave服务),已达到动态地设置子服务器的目的。配置文件格式

    <slave_config>
       <masters>
       <slaveserver>
         <name>master1</name>
         <hostname>localhost</hostname>
         <port>8080</port>
         <username>cluster</username>
         <password>cluster</password>
         <master>Y</master>
       </slaveserver>
      </masters>
      <report_to_masters>Y</report_to_masters>
    <slaveserver>
      <name>slave4-8084</name>
      <hostname>localhost</hostname>
      <port>8084</port>
      <username>cluster</username>
      <password>cluster</password>
      <master>N</master>
    </slaveserver>
    </slave_config>

    这个配置文件主要包括以下几个节点

    Ø       masters: 这里列出来的服务器是当前子服务器需要向其汇报状态的主服务器。如果当前这个子服务器是主服务器,则它将连接其它的主服务器来获得这个集群中的所有子服务器。

    Ø       report_to_masters : 如果为Y,则表示需要向定义的主服务器发送消息以表明该从属服务器存在

    Ø       slaveserver : 这里定义的就是当前carte实例运行时需要的子服务器的配置情况

    这里定义的username和password在向主服务器调用Register服务时连接主服务器时提供的安全设置。在 <slaveserver>部分,你可以使用<network_interface> 参数,这个参数的优先级高于<hostname>参数 ,如果你的机器中安装有多个网卡,这个设置可以起作用。

    程序启动

    Kettle提供了org.pentaho.di.www.Carte类,你可以通过该类提供的函数来启动或者停止子服务器。

    Ø         启动子服务器

    SlaveServerConfig config = new SlaveServerConfig(hostname, port, false);

    Carte. runCarte(config);

    Ø         停止子服务器

    carte.getWebServer().stopServer();

    子服务器内幕

    我们前面提到过子服务器实际上就是一个web server,该web server是基于Jetty这个嵌入式的开源servlet容器。

    这个web server主要是提供转换运行的环境,另外一个重要的功能通过提供servlet来在客户端、主服务器和从属服务器之间进行通讯和控制。主服务器和从属服务器之间是通过httpClient来进行通讯的,通讯时传递的数据是xml格式。通过提供的servlet,可以实现启动、停止、暂停转换或者作业、获得转换或者作业的状态、注册子服务器、获得子服务器的列表等等

    Kettle主要提供了以下的几种基于servlet的服务

    Ø         GetRootServlet:获得Carte的根目录

    Ø         GetStatusServlet:获得在服务器上运行的所有的转换和作业的状态

    Ø         GetTransStatusServlet:获得在服务器上运行的某个指定的转换的每个步骤的运行状态。

    Ø         PrepareExecutionTransServlet:让服务器上的某个指定的转换做好运行的准备。

    Ø         StartTransServlet:执行服务器上的某个指定的转换

    Ø         PauseTransServlet:暂停或者重新运行某一个转换

    Ø         StopTransServlet:停止正在运行的转换

    Ø         CleanupTransServlet:清理运行转换时的环境

    Ø         AddTransServlet:向子服务器中增加某个转换。如果服务器中有正在运行或者准备运行的相同名字的转换,则抛出异常。

    Ø         AllocateServerSocketServlet:分配一个新的socket端口号。这个端口号是基于你在定义cluster schema中设置的端口号,依次加1

    Ø         StartJobServlet:执行服务器上某个指定的作业

    Ø         StopJobServlet:停止正在运行的作业

    Ø         GetJobStatusServlet:获得某个指定作业的状态

    Ø         AddJobServlet:向当前的子服务器中添加某个作业。

    Ø         RegisterSlaveServlet:注册某个服务器的信息。服务器信息包括子服务器是否活动、最新活动的时间、最新不活动的时间。这个在dynamic cluster中需要用到,由从属服务器向主服务器汇报当前状态。

    Ø         GetSlavesServlet:获得集群中子服务器的信息

    Ø         AddExportServlet:以zip文件的形式向caret服务器传递作业或者转换信息,并将信息加入到服务器中。

    运行转换

    在spoon中运行

    在kettle的集成设计环境spoon中,你可以选择转换中的“运行”菜单项,或者按F9快捷键,弹出以下的窗口

    这里有三个选项来决定转换是以什么方式来执行

    Ø       本地执行: 转换或者作业将在你现在使用的JVM中运行。

    Ø       远程执行: 允许你指定一个想运行转换的远程服务器。这需要你在远程服务器上安装Pentaho Data Integration(Kettle)并且运行Carte子服务器。

    Ø       集群方式执行: 允许你在集群环境下执行作业或者转换

     

    当你选择“集群方式执行”选项是,你可以选择以下的选项

    Ø       提交转换: 分解转换并且将转换提交到不同的主服务器和从属服务器。

    Ø       准备执行: 它将在主服务器和从属服务器上执行转换的初始化阶段。

    Ø       开始执行: 它将在主服务器和从属服务器中执行实际的转换任务。

    Ø       显示转换: 显示将要在集群上执行的生成的转换(可以参看下面的分析).

    编程运行

    你也可以通过使用Kettle提供的API通过编程来以集群的方式运行转换。

    TransMeta transMeta = new TransMeta("cluster.ktr");

    //设置执行模式

    TransExecutionConfiguration config = new TransExecutionConfiguration();

    config.setExecutingClustered(true);

    config.setExecutingLocally(false);

    config.setExecutingRemotely(false);

    config.setClusterPosting(true);

    config.setClusterPreparing(true);

    config.setClusterStarting(true);

    config.setLogLevel(LogWriter.LOG_LEVEL_BASIC);

    TransSplitter transSplitter = Trans.executeClustered(transMeta, config);

    long nrErrors = Trans.monitorClusteredTransformation("cluster  test", transSplitter, null, 1);

    需要注意的是这段代码可以在一个独立的JVM中执行,而不必要在主服务器中执行。

    运行内幕

    当以集群方式来运行转换时,Kettle主要执行以下几个步骤来执行分布式的处理

    Ø 分解转换

    在定义转换的时候,如果在某个步骤中定义使用集群,那么这个步骤其实是在从属服务器(slave server)上执行的,例如我们在前面定义转换的modify javascript value步骤中,我们定义了使用集群,那么这个步骤将在从属服务器中执行;而Cx4表示这个步骤是在4个从属服务器上执行。如果步骤中没有定义集群,则表示该步骤是在主服务器(master server)上执行。如果前一步骤在主服务器上执行,而后一步骤需要在从属服务器上执行,或者相反,则这时需要分别在前一步骤和后一步骤之间建立一个remoteStep步骤,前面的remoteStep建立socketWriter进程,它负责从上一步骤中取出数据然后通过socket传输到对应的子服务器的remoteStep中。而后一步骤所在的子服务器的remoteStep步骤则建立一个socketReader,负责从socket中获取数据,并将数据将数据传输到后一步骤中,以供后一步骤来进行后续处理。

    所以在以集群方式执行转换时,首要的任务是将转换分解成可以在各个子服务器上执行的转换。

    我们还是以上面建立的转换来进行分析描述:

    上图是在主服务器上建立的转换

    And 4 slaves transformations:

    上图是在4个从属服务器上建立的转换,我们可以注意到这四个从属服务器上的转换是一样的,除了端口号不一样。另外我们还注意到在前述Cluster Schema 定义中我们指定了端口号为4000,则为每一个建立的socket连接就是端口号4000开始,依次加1。另外,还可以看到数据是通过使用socket  Writer和socket Reader的remoteStep步骤通过TCP/IP的socket来传递数据的。

    Ø         提交转换

    对于第一步骤生成的子转换,将调用每个子服务器提供的AddTransServlet服务将转换的信息增加到每个子服务器中(包括主服务器和从属服务器)。

    Ø         准备转换

    调用每个子服务器的PrepareExecutionTransServlet服务来准备转换

    Ø         启动转换

    调用每个子服务器的StartExecutionTransServlet服务来启动转换。

    Ø         监控转换

    在各服务器的转换都启动后,调用Trans.monitorClusteredTransformation来监控各个服务器的运行状态(使用各子服务器提供的GetTransStatusServlet服务来获得每个子服务器的状态)。

    例子

    目的

    做一个转换(表输入---à排序--à表输出)

    然后在两台pc机器上实验。把集群放到排序插件上。

    配置两台子服务器

    创建子服务器

    在主对象下的转换下的子服务器右键单击新建。

    右键单击子服务器新建

    填写相关的配置,用户名和密码为cluster,如果要修改得修改kettle默认路径下的pwd下的kettle.pwd文件里的用户名密码。

    这个是从属服务器。

    配置schemas

    新建schemas

    在选择子服务器中选择这两个服务器。

    开启两台机器的carte服务

    在10.2.4.81机器和10.2.4.188机器的控制台开启carte服务。

    C:/pdi-open-3.1.0-826是保存kettle的文件夹

    188机器也和他一样。

    在转换中添加集群

    右击字段选择选择集群。

    点击确定。

    出现cx1代表成功。

    然后运行,就OK了。

    ---------------------------------以下内容是我自己添加的----------------------------------------

    更多资料:

    http://forums.pentaho.com/showthread.php?60500-clustering-in-Kettle
    http://forums.pentaho.com/archive/index.php/t-54398.html

    全文
2019-10-30 10:51:20