小贝子编程

如何在桑扎写我自己的工作



最近我试图在Samza框架上做一些流处理工作。我已经成功地部署了hello-samza示例。然而，当我试图写自己的工作时，我不知道从哪里开始我的工作。

我读过这份文件，但还是不明白要点。有人能帮我吗：

  1. 我的代码的体系结构是什么（源代码、库代码和配置）
  2. 我的代码将被推入哪个目录
  3. 我还需要做什么其他工作来运行我的代码

你的建议会帮我很多，非常感谢！

构建自己的Jobs非常简单。首先得到你好samza:

git clone https://git.apache.org/samza-hello-samza.git hello-samza

下一步是通过以下命令设置系统：

bin/grid bootstrap

请确保jps 一切正常

下一步是从pom.xml中删除apacherat插件，而不是在hello-samza中构建项目。

当你远程调用时，你可以在src文件夹（MyTask.java）中添加一个java文件Job，也可以在配置目录（My.Task.properties）中添加.properties文件

这是一个空作业示例（MyTask.java）

package com.samza;
public class MyTask implements StreamTask {
    private static final SystemStream OUTPUT_STREAM = new SystemStream("kafka","topicOut"); 
    public void process (IncomingMessageEnvelope envelope,                  MessageCollector collector,                 
    TaskCoordinator coordinator) throws Exception { 
        // Do something useful
    }
}

不要忘记实现.properties文件。

如果你有一个无错误的代码，用maven构建，比如：

mvn clean package
mkdir -p deploy/samza
tar -xvf ./samza-job-package/target/samza-job-package-0.10.0-dist.tar.gz -C deploy/samza

之后，您的服务器启动（如果不是，您可以通过./bin/grid start all)启动，您可以使用deploy/samza/bin/run-job.sh --config-factory=org.apache.samza.config.factories.PropertiesConfigFactory --config-path=file://$PWD/deploy/samza/config/MyTask.properties 部署您的作业

并由kafka客户端消费者消费结果deploy/kafka/bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic outTopic

如果按照Hello-Samza的说明操作，您将在本地计算机上运行一个功能齐全的Zookeeper、Kafka和Yarn/Samza集群。有了这个项目，你就可以运行与维基百科提要相关的任务来测试东西。

然而，和您一样，我在为新任务提供正确的目录结构和构建设置时遇到了一些困难（没有集群管理的东西）。因此，我创建了hello-samza基础，去掉了hello-samza之外的新任务所不需要的一切。我在自述中包含了关于构建新任务的说明。

就部署而言，这有点复杂。读一些关于创建Zookeeper、Kafka和Yarn集群的文章。

我通过Maven Eclipse项目创建了Samza作业。0.9.2版本的依赖项与此内容一起加载在pom.xml文件中（我有一些版本问题，所以您可能在那里有一些工作）：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
      <modelVersion>4.0.0</modelVersion>
      <groupId>com.acio.samza</groupId>
      <artifactId>samzafroga</artifactId>
      <version>0.0.1</version>
      <name>samzafroga</name>
       <dependencies>
        <dependency>
          <groupId>org.apache.samza</groupId>
          <artifactId>samza-api</artifactId>
        </dependency>
        <dependency>
          <groupId>org.apache.samza</groupId>
          <artifactId>samza-core_2.10</artifactId>
        </dependency>
        <dependency>
          <groupId>org.apache.samza</groupId>
          <artifactId>samza-log4j</artifactId>
        </dependency>
        <dependency>
          <groupId>org.apache.samza</groupId>
          <artifactId>samza-shell</artifactId>
        </dependency>
        <dependency>
          <groupId>org.apache.samza</groupId>
          <artifactId>samza-yarn_2.10</artifactId>
          <exclusions>
            <exclusion>
                <artifactId>jdk.tools</artifactId>
                <groupId>jdk.tools</groupId>
            </exclusion>
        </exclusions>
        </dependency>
        <dependency>
          <groupId>org.apache.samza</groupId>
          <artifactId>samza-kv_2.10</artifactId>
        </dependency>
        <dependency>
          <groupId>org.apache.samza</groupId>
          <artifactId>samza-kv-rocksdb_2.10</artifactId>
        </dependency>
        <dependency>
          <groupId>org.apache.samza</groupId>
          <artifactId>samza-kafka_2.10</artifactId>
        </dependency>
        <dependency>
          <groupId>org.apache.kafka</groupId>
          <artifactId>kafka_2.10</artifactId>
        </dependency>
        <dependency>
          <groupId>org.schwering</groupId>
          <artifactId>irclib</artifactId>
        </dependency>
        <dependency>
          <groupId>org.slf4j</groupId>
          <artifactId>slf4j-api</artifactId>
        </dependency>
        <dependency>
          <groupId>org.slf4j</groupId>
          <artifactId>slf4j-log4j12</artifactId>
        </dependency>
        <dependency>
          <groupId>org.codehaus.jackson</groupId>
          <artifactId>jackson-jaxrs</artifactId>
        </dependency>
        <dependency>
          <groupId>org.apache.hadoop</groupId>
          <artifactId>hadoop-hdfs</artifactId>
        </dependency>
        <dependency>
            <groupId>org.eclipse.jetty</groupId>
            <artifactId>jetty-server</artifactId>
            <version>${jettyVersion}</version>
        </dependency>
        <dependency>
            <groupId>org.eclipse.jetty</groupId>
            <artifactId>jetty-webapp</artifactId>
            <version>${jettyVersion}</version>
        </dependency>
       </dependencies>
      <properties>
        <!-- maven specific properties -->
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <samza.version>0.8.0</samza.version>
        <jettyVersion>7.6.16.v20140903</jettyVersion>
      </properties>
        <repositories>
        <repository>
          <id>apache-releases</id>
          <url>https://repository.apache.org/content/groups/public</url>
        </repository>
        <repository>
          <id>scala-tools.org</id>
          <name>Scala-tools Maven2 Repository</name>
          <url>https://oss.sonatype.org/content/groups/scala-tools</url>
        </repository>
      </repositories>
      <pluginRepositories>
        <pluginRepository>
          <id>scala-tools.org</id>
          <name>Scala-tools Maven2 Repository</name>
          <url>http://scala-tools.org/repo-releases</url>
        </pluginRepository>
      </pluginRepositories>
       <build>
        <pluginManagement>
          <plugins>
            <plugin>
              <groupId>org.apache.rat</groupId>
              <artifactId>apache-rat-plugin</artifactId>
              <version>0.9</version>
              <configuration>
                <excludes>
                  <exclude>*.patch</exclude>
                  <exclude>**/target/**</exclude>
                  <exclude>*.json</exclude>
                  <exclude>.vagrant/**</exclude>
                  <exclude>.git/**</exclude>
                  <exclude>*.md</exclude>
                  <exclude>docs/**</exclude>
                  <exclude>config/**</exclude>
                  <exclude>bin/**</exclude>
                  <exclude>.gitignore</exclude>
                  <exclude>**/.cache/**</exclude>
                  <exclude>deploy/**</exclude>
                  <exclude>**/.project</exclude>
                </excludes>
              </configuration>
             </plugin>
          </plugins>
        </pluginManagement>
        <plugins>
          <!-- plugin to build the tar.gz file filled with examples -->
          <plugin>
            <artifactId>maven-assembly-plugin</artifactId>
            <version>2.3</version>
            <configuration>
              <descriptors>
                <descriptor>src/assembly/bin.xml</descriptor>
              </descriptors>
               <descriptorRefs>
                <descriptorRef>jar-with-dependencies</descriptorRef>
              </descriptorRefs>
            </configuration>
            <executions>
              <execution>
                <id>make-assembly</id>
                <phase>package</phase>
                <goals>
                  <goal>single</goal>
                </goals>
              </execution>
            </executions>
          </plugin>
        </plugins>
      </build>
       <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.apache.samza</groupId>
                <artifactId>samza-api</artifactId>
                <version>${samza.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.samza</groupId>
                <artifactId>samza-core_2.10</artifactId>
                <version>${samza.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.samza</groupId>
                <artifactId>samza-log4j</artifactId>
                <version>${samza.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.samza</groupId>
                <artifactId>samza-shell</artifactId>
                <version>${samza.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.samza</groupId>
                <artifactId>samza-yarn_2.10</artifactId>
                <version>${samza.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.samza</groupId>
                <artifactId>samza-kv_2.10</artifactId>
                <version>${samza.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.samza</groupId>
                <artifactId>samza-kv-rocksdb_2.10</artifactId>
                <version>${samza.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.samza</groupId>
                <artifactId>samza-kafka_2.10</artifactId>
                <version>${samza.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka_2.10</artifactId>
                <version>0.8.2.0</version>
            </dependency>
            <dependency>
                <groupId>org.schwering</groupId>
                <artifactId>irclib</artifactId>
                <version>1.10</version>
            </dependency>
            <dependency>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-api</artifactId>
                <version>1.6.2</version>
            </dependency>
            <dependency>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-log4j12</artifactId>
                <version>1.6.2</version>
            </dependency>
            <dependency>
                <groupId>org.codehaus.jackson</groupId>
                <artifactId>jackson-jaxrs</artifactId>
                <version>1.8.5</version>
            </dependency>
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-hdfs</artifactId>
                <version>2.6.0</version>
            </dependency>
        </dependencies>
       </dependencyManagement>
    </project>

作业的基本代码是这样的：

package xxxx;
import java.util.Map;
import org.apache.samza.config.Config;
import org.apache.samza.system.IncomingMessageEnvelope;
import org.apache.samza.system.OutgoingMessageEnvelope;
import org.apache.samza.system.SystemStream;
import org.apache.samza.task.MessageCollector;
import org.apache.samza.task.StreamTask;
import org.apache.samza.task.TaskContext;
import org.apache.samza.task.TaskCoordinator;
public class Redirect implements StreamTask {
      private final SystemStream OUTPUT_STREAM = new SystemStream("kafka", "samzaout");
     public void process(IncomingMessageEnvelope envelope,
         MessageCollector collector,
         TaskCoordinator coordinator)
     {
         String msg = (String)envelope.getMessage();
         // Transformation
         String outmsg = "xxx-" + msg + "-xxx";
         collector.send(new OutgoingMessageEnvelope(OUTPUT_STREAM, outmsg));
     }
}

编译后，您需要将其分组到一个jar文件中，并将其放置在所有samza节点、web或hdfs都可以访问的位置。

从您必须创建的属性文件中引用它才能启动它。在porject网页中查找示例。

阅读更多文档，查看更多hello-samza示例，如果您将其部署到YARN，请阅读更多关于它的信息。你正在寻找的所有答案都在那里。

你好桑扎有三份工作。选择一个并遵循它，配置，启动脚本等

以下是来自hello-samza页面的如何启动维基百科提要作业

deploy/samza/bin/run-job.sh --config-factory=org.apache.samza.config.factories.PropertiesConfigFactory --config-path=file://$PWD/deploy/samza/config/wikipedia-feed.properties

属性文件显示编译后的作业/任务代码以及其他内容的位置。维基百科提要作业/任务的源代码如下：

https://github.com/apache/samza-hello-samza/blob/master/src/main/java/samza/examples/wikipedia/task/WikipediaFeedStreamTask.java

只需修改此作业，或者复制并修改，就可以让您的作业继续运行。

相关内容

  • 没有找到相关文章

最新更新


热门标签：

javascript python java c# php android html jquery c++ css ios sql mysql arrays asp.net json python-3.x ruby-on-rails .net sql-server django objective-c excel regex ruby linux ajax iphone xml vba spring asp.net-mvc database wordpress string postgresql wpf windows xcode bash git oracle list vb.net multithreading eclipse algorithm macos powershell visual-studio image forms numpy scala function api selenium