干货BitSailConnector开发详解系列一Source

public interface Source<T, SplitT extends SourceSplit, StateT extends Serializable> extends Serializable, TypeInfoConverterFactory {
    /**
     * Run in client side for source initialize;
     */
    void configure(ExecutionEnviron execution, BitSailConfiguration readerConfiguration) throws IOException;

    /**
     * Indicate the Source type.
     */
    Boundedness getSourceBoundedness();

    /**
     * Create Source Reader.
     */
    SourceReader<T, SplitT> createReader(SourceReader.Context readerContext);

    /**
     * Create split coordinator.
     */
    SourceSplitCoordinator<SplitT, StateT> createSplitCoordinator(SourceSplitCoordinator.Context<SplitT, StateT> coordinatorContext);

    /**
     * Get Split serializer for the framework, {@link SplitT} should implement from {@link Serializable}
     */
    default BinarySerializer<SplitT> getSplitSerializer() {
        return new SimpleBinarySerializer<>();
    }

    /**
     * Get State serializer for the framework, {@link StateT} should implement from {@link Serializable}
     */
    default BinarySerializer<StateT> getSplitCoordinatorCheckpointSerializer() {
        return new SimpleBinarySerializer<>();
    }

    /**
     * Create type info converter for the source, default value {@link BitSailTypeInfoConverter}
     */
    default TypeInfoConverter createTypeInfoConverter() {
        return new BitSailTypeInfoConverter();
    }

    /**
     * Get Source' name.
     */
    String getReaderName();
}

@Override
public void configure(ExecutionEnviron execution, BitSailConfiguration readerConfiguration) {
    this.readerConfiguration = readerConfiguration;
    this.commonConfiguration = execution.getCommonConfiguration();
}

@Override
public Boundedness getSourceBoundedness() {
    return Mode.BATCH.equals(Mode.getJobRunMode(commonConfiguration.get(CommonOptions.JOB_TYPE))) ?
            Boundedness.BOUNDEDNESS :
            Boundedness.UNBOUNDEDNESS;
}

public Boundedness getSourceBoundedness() {
    return Boundedness.BOUNDEDNESS;
}

@Override
public TypeInfoConverter createTypeInfoConverter() {
    return new FileMappingTypeInfoConverter(getReaderName());
}

# Clickhouse Type to BitSail Type
engine.type.to.bitsail.type.converter:
  - source.type: int32
    target.type: int
  - source.type: float64
    target.type: double
  - source.type: string
    target.type: string
  - source.type: date
    target.type: date.date
  - source.type: null
    target.type: void

# BitSail Type to Clickhouse Type
bitsail.type.to.engine.type.converter:
  - source.type: int
    target.type: int32
  - source.type: double
    target.type: float64
  - source.type: date.date
    target.type: date
  - source.type: string
    target.type: string

"reader": {
  "class": "com.bytedance.bitsail.connector.clickhouse.source.ClickhouseSource",
  "jdbc_url": "jdbc:clickhouse://localhost:8123",
  "db_name": "default",
  "table_name": "test_ch_table",
  "split_field": "id",
  "split_config": "{"name": "id", "lower_bound": 0, "upper_bound": "10000", "split_num": 3}",
  "sql_filter": "( id % 2 == 0 )",
  "columns": [
    {
      "name": "id",
      "type": "int64"
    },
    {
      "name": "int_type",
      "type": "int32"
    },
    {
      "name": "double_type",
      "type": "float64"
    },
    {
      "name": "string_type",
      "type": "string"
    },
    {
      "name": "p_date",
      "type": "date"
    }
  ]
}

@Override
public TypeInfoConverter createTypeInfoConverter() {
    return new BitSailTypeInfoConverter();
}

"reader": {
  "class": "com.bytedance.bitsail.connector.hadoop.source.HadoopSource",
  "path_list": "hdfs://127.0.0.1:9000/test_namespace/source/test.json",
  "content_type": "json",
  "reader_parallelism_num": 1,
  "columns": [
    {
      "name": "id",
      "type": "int"
    },
    {
      "name": "string_type",
      "type": "string"
    },
    {
      "name": "map_string_string",
      "type": "map<string,string>"
    },
    {
      "name": "array_string",
      "type": "list<string>"
    }
  ]
}

public SourceReader<Row, RocketMQSplit> createReader(SourceReader.Context readerContext) {
    return new RocketMQSourceReader(
        readerConfiguration,
        readerContext,
        getSourceBoundedness()
    );
}

public SourceSplitCoordinator<RocketMQSplit, RocketMQState> createSplitCoordinator(SourceSplitCoordinator.Context<RocketMQSplit, RocketMQState> coordinatorContext) {
    return new RocketMQSourceSplitCoordinator(
        coordinatorContext,
        readerConfiguration,
        getSourceBoundedness()
    );
}

public interface ParallelismComputable extends Serializable {
    /**
     * give a parallelism advice for reader/writer based on configurations and upstream parallelism advice
     *
     * @param commonConf     common configuration
     * @param selfConf       reader/writer configuration
     * @param upstreamAdvice parallelism advice from upstream (when an operator has no upstream in DAG, its upstream is
     *                       global parallelism)
     * @return parallelism advice for the reader/writer
     */
    ParallelismAdvice getParallelismAdvice(BitSailConfiguration commonConf,
                                           BitSailConfiguration selfConf,
                                           ParallelismAdvice upstreamAdvice) throws Exception;
}

public ParallelismAdvice getParallelismAdvice(BitSailConfiguration commonConfiguration,
                                              BitSailConfiguration rocketmqConfiguration,
                                              ParallelismAdvice upstreamAdvice) throws Exception {
    String cluster = rocketmqConfiguration.get(RocketMQSourceOptions.CLUSTER);
    String topic = rocketmqConfiguration.get(RocketMQSourceOptions.TOPIC);
    String consumerGroup = rocketmqConfiguration.get(RocketMQSourceOptions.CONSUMER_GROUP);
    DefaultLitePullConsumer consumer = RocketMQUtils.prepareRocketMQConsumer(rocketmqConfiguration, String.format(SOURCE_INSTANCE_NAME_TEMPLATE,
        cluster,
        topic,
        consumerGroup,
        UUID.randomUUID()
    ));
    try {
        consumer.start();
        Collection<MessageQueue> messageQueues = consumer.fetchMessageQueues(topic);
        int adviceParallelism = Math.max(CollectionUtils.size(messageQueues) / DEFAULT_ROCKETMQ_PARALLELISM_THRESHOLD, 1);

        return ParallelismAdvice.builder()
            .adviceParallelism(adviceParallelism)
            .enforceDownStreamChain(true)
            .build();
    } finally {
        consumer.shutdown();
    }
}