Mailing-List: contact dev-help@samoa.incubator.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@samoa.incubator.apache.org
From: gdfm <git@git.apache.org>
To: dev@samoa.incubator.apache.org
Reply-To: dev@samoa.incubator.apache.org
References: <git-pr-32-incubator-samoa@git.apache.org>
In-Reply-To: <git-pr-32-incubator-samoa@git.apache.org>
Subject: [GitHub] incubator-samoa pull request: SAMOA-40: Add Kafka stream
 reader mo...
Content-Type: text/plain
Message-Id: <20150721110714.F369BDFFB9@git1-us-west.apache.org>
Date: Tue, 21 Jul 2015 11:07:14 +0000 (UTC)

Github user gdfm commented on a diff in the pull request:

    https://github.com/apache/incubator-samoa/pull/32#discussion_r35091207
  
    --- Diff: samoa-api/src/main/java/org/apache/samoa/streams/kafka/KafkaStream.java ---
    @@ -0,0 +1,201 @@
    +package org.apache.samoa.streams.kafka;
    +
    +/*
    + * #%L
    + * SAMOA
    + * %%
    + * Copyright (C) 2014 - 2015 Apache Software Foundation
    + * %%
    + * Licensed under the Apache License, Version 2.0 (the "License");
    + * you may not use this file except in compliance with the License.
    + * You may obtain a copy of the License at
    + * 
    + *      http://www.apache.org/licenses/LICENSE-2.0
    + * 
    + * Unless required by applicable law or agreed to in writing, software
    + * distributed under the License is distributed on an "AS IS" BASIS,
    + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    + * See the License for the specific language governing permissions and
    + * limitations under the License.
    + * #L%
    + */
    +
    +import com.github.javacliparser.IntOption;
    +import com.github.javacliparser.StringOption;
    +import org.apache.samoa.instances.*;
    +import org.apache.samoa.moa.core.Example;
    +import org.apache.samoa.moa.core.FastVector;
    +import org.apache.samoa.moa.core.InstanceExample;
    +import org.apache.samoa.moa.core.ObjectRepository;
    +import org.apache.samoa.moa.options.AbstractOptionHandler;
    +import org.apache.samoa.moa.streams.InstanceStream;
    +import org.apache.samoa.moa.tasks.TaskMonitor;
    +import org.slf4j.Logger;
    +import org.slf4j.LoggerFactory;
    +
    +import java.util.ArrayList;
    +import java.util.LinkedList;
    +import java.util.List;
    +import java.util.Queue;
    +
    +public class KafkaStream extends AbstractOptionHandler implements
    +        InstanceStream {
    +
    +    private static final long serialVersionUID = 1L;
    +
    +    protected InstancesHeader streamHeader;
    +
    +    protected Instances instances;
    +
    +    private KafkaReader reader;
    +
    +    protected InstanceExample lastInstanceRead;
    +
    +    protected Queue<String> instanceQueue;
    +
    +    private static final Logger logger = LoggerFactory
    +            .getLogger(KafkaStream.class);
    +
    +    public void KafkaReader() {
    +        reader = new KafkaReader();
    +    }
    +
    +    public IntOption classIndexOption = new IntOption("classIndex", 'c',
    +            "Class index of data. 0 for none or -1 for last attribute in file.",
    +            -1, -1, Integer.MAX_VALUE);
    +
    +    public IntOption numAttrOption = new IntOption("numNumerics", 'u',
    +            "The number of numeric attributes in" +
    +                    " dataset", 300, 0,
    +            Integer.MAX_VALUE);
    +
    +    public StringOption topicOption = new StringOption("topic", 't',
    +            "Topic in the kafka to be used for reading data", "test");
    +
    +    public IntOption numMaxreadOption = new IntOption("numMaxread", 'r',
    +            "Number of instances to be read in single read from kafka", 1, 0,
    +            Integer.MAX_VALUE);
    +
    +    public IntOption partitionOption = new IntOption("partition", 'n',
    +            "Partition number to be used for reading data", 0);
    +
    +    public IntOption portOption = new IntOption("port", 'p',
    +            "Port in kafka to read data", 9092);
    +
    +    public StringOption seedOption = new StringOption("seed", 's',
    +            "Seeds for kafka", "localhost");
    +
    +    public IntOption numClassesOption = new IntOption("numClasses", 'k',
    +            "The number of classes in the data.", 2, 2, Integer.MAX_VALUE);
    +
    +    public IntOption timeDelayOption = new IntOption("timeDelay", 'y',
    +            "Time delay in milliseconds between two read from kafka", 0, 0, Integer.MAX_VALUE);
    +
    +    @Override
    +    protected void prepareForUseImpl(TaskMonitor monitor,
    +                                     ObjectRepository repository) {
    +        this.reader = new KafkaReader();
    +        generateHeader();
    +        instanceQueue = new LinkedList<String>();
    +        this.lastInstanceRead = null;
    +    }
    +
    +    protected void generateHeader() {
    --- End diff --
    
    I am not sure about the design of this class, especially of this method.
    Would you care to explain the rationale behind it?


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---