[jira] [Commented] (CARBONDATA-298) 3. Add InputProcessorStep which should iterate recordreader and parse the data as per the data type.

[Akash R Nilugal (Jira)]

    [ https://issues.apache.org/jira/browse/CARBONDATA-298?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15591236#comment-15591236 ]

ASF GitHub Bot commented on CARBONDATA-298:
-------------------------------------------

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

    https://github.com/apache/incubator-carbondata/pull/240#discussion_r84232604
 
    --- Diff: processing/src/main/java/org/apache/carbondata/processing/newflow/steps/input/InputProcessorStepImpl.java ---
    @@ -0,0 +1,171 @@
    +package org.apache.carbondata.processing.newflow.steps.input;
    +
    +import java.util.ArrayList;
    +import java.util.Iterator;
    +import java.util.List;
    +
    +import org.apache.carbondata.common.CarbonIterator;
    +import org.apache.carbondata.common.logging.LogService;
    +import org.apache.carbondata.common.logging.LogServiceFactory;
    +import org.apache.carbondata.core.constants.CarbonCommonConstants;
    +import org.apache.carbondata.core.util.CarbonProperties;
    +import org.apache.carbondata.processing.newflow.AbstractDataLoadProcessorStep;
    +import org.apache.carbondata.processing.newflow.CarbonDataLoadConfiguration;
    +import org.apache.carbondata.processing.newflow.DataField;
    +import org.apache.carbondata.processing.newflow.constants.DataLoadProcessorConstants;
    +import org.apache.carbondata.processing.newflow.exception.CarbonDataLoadingException;
    +import org.apache.carbondata.processing.newflow.parser.CarbonParserFactory;
    +import org.apache.carbondata.processing.newflow.parser.GenericParser;
    +import org.apache.carbondata.processing.newflow.row.CarbonRow;
    +import org.apache.carbondata.processing.newflow.row.CarbonRowBatch;
    +
    +/**
    + * It reads data from record reader and sends data to next step.
    + */
    +public class InputProcessorStepImpl extends AbstractDataLoadProcessorStep {
    +
    +  private static final LogService LOGGER =
    +      LogServiceFactory.getLogService(InputProcessorStepImpl.class.getName());
    +
    +  private GenericParser[] genericParsers;
    +
    +  private List<Iterator<Object[]>> inputIterators;
    +
    +  public InputProcessorStepImpl(CarbonDataLoadConfiguration configuration,
    +      AbstractDataLoadProcessorStep child, List<Iterator<Object[]>> inputIterators) {
    +    super(configuration, child);
    +    this.inputIterators = inputIterators;
    +  }
    +
    +  @Override public DataField[] getOutput() {
    +    DataField[] fields = configuration.getDataFields();
    +    String[] header = configuration.getHeader();
    +    DataField[] output = new DataField[fields.length];
    +    int k = 0;
    +    for (int i = 0; i < header.length; i++) {
    +      for (int j = 0; j < fields.length; j++) {
    +        if (header[j].equalsIgnoreCase(fields[j].getColumn().getColName())) {
    +          output[k++] = fields[j];
    +          break;
    +        }
    +      }
    +    }
    +    return output;
    +  }
    +
    +  @Override public void initialize() throws CarbonDataLoadingException {
    +    DataField[] output = getOutput();
    +    genericParsers = new GenericParser[output.length];
    +    for (int i = 0; i < genericParsers.length; i++) {
    +      genericParsers[i] = CarbonParserFactory.createParser(output[i].getColumn(),
    +          (String[]) configuration
    +              .getDataLoadProperty(DataLoadProcessorConstants.COMPLEX_DELIMITERS));
    +    }
    +  }
    +
    +  private int getNumberOfCores() {
    +    int numberOfCores;
    +    try {
    +      numberOfCores = Integer.parseInt(CarbonProperties.getInstance()
    +          .getProperty(CarbonCommonConstants.NUM_CORES_LOADING,
    +              CarbonCommonConstants.NUM_CORES_DEFAULT_VAL));
    +    } catch (NumberFormatException exc) {
    +      numberOfCores = Integer.parseInt(CarbonCommonConstants.NUM_CORES_DEFAULT_VAL);
    +    }
    +    return numberOfCores;
    +  }
    +
    +  private int getBatchSize() {
    +    int batchSize;
    +    try {
    +      batchSize = Integer.parseInt(configuration
    +          .getDataLoadProperty(DataLoadProcessorConstants.DATA_LOAD_BATCH_SIZE,
    +              DataLoadProcessorConstants.DATA_LOAD_BATCH_SIZE_DEFAULT).toString());
    +    } catch (NumberFormatException exc) {
    +      batchSize = Integer.parseInt(DataLoadProcessorConstants.DATA_LOAD_BATCH_SIZE_DEFAULT);
    +    }
    +    return batchSize;
    +  }
    +
    +  @Override public Iterator<CarbonRowBatch>[] execute() {
    +    int batchSize = getBatchSize();
    +    List<Iterator<Object[]>>[] readerIterators = partitionInputReaderIterators();
    +    Iterator<CarbonRowBatch>[] outIterators = new Iterator[readerIterators.length];
    +    for (int i = 0; i < outIterators.length; i++) {
    +      outIterators[i] = new InputProcessorIterator(readerIterators[i], genericParsers, batchSize);
    +    }
    +    return outIterators;
    +  }
    +
    +  private List<Iterator<Object[]>>[] partitionInputReaderIterators() {
    +    int numberOfCores = getNumberOfCores();
    +    if (inputIterators.size() < numberOfCores) {
    +      numberOfCores = inputIterators.size();
    --- End diff --
   
    please use a different name (suggest parallelism), the meaning of this variable is `min(inputIterator.size(), numberOfCores)`

> 3. Add InputProcessorStep which should iterate recordreader and parse the data as per the data type.
> ----------------------------------------------------------------------------------------------------
>
>                 Key: CARBONDATA-298
>                 URL: https://issues.apache.org/jira/browse/CARBONDATA-298
>             Project: CarbonData
>          Issue Type: Sub-task
>            Reporter: Ravindra Pesala
>             Fix For: 0.2.0-incubating
>
>
> Add InputProcessorStep which should iterate recordreader/RecordBufferedWriter and parse the data as per the data types.

--
This message was sent by Atlassian JIRA
(v6.3.4#6332)