Add streaming support

[jaymo001]

Add streaming support for anchored features.
Example:

1. Define streaming input source on Kafka
   Define input data schema:

 schema = AvroJsonSchema(schemaStr="""
    {
        "type": "record",
        "name": "DriverTrips",
        "fields": [
            {"name": "driver_id", "type": "long"},
            {"name": "trips_today", "type": "int"},
            {
                "name": "datetime",
                "type": {"type": "long", "logicalType": "timestamp-micros"}
            }
        ]
    }
    """)
stream_source = KafKaSource(name="kafkaStreamingSource",
                              kafkaConfig=KafkaConfig(brokers=["feathrazureci.servicebus.windows.net:9093"],
                                                      topics=["feathrcieventhub"],
                                                      schema=schema)
                              )

3. Define feature definition with the Kafka source

  driver_id = TypedKey(key_column="driver_id",
                          key_column_type=ValueType.INT64,
                          description="driver id",
                          full_name="nyc driver id")

    kafkaAnchor = FeatureAnchor(name="kafkaAnchor",
                                      source=stream_source,
                                      features=[Feature(name="f_modified_streaming_count",
                                                        feature_type=INT32,
                                                        transform="trips_today + 1",
                                                        key=driver_id),
                                                Feature(name="f_modified_streaming_count2",
                                                        feature_type=INT32,
                                                        transform="trips_today + 2",
                                                        key=driver_id)]
                                      )

5. Start streaming job

 redisSink = RedisSink(table_name="kafkaSampleDemoFeature", streaming=True, streamingTimeoutMs=10000)
    settings = MaterializationSettings(name="kafkaSampleDemo",
                                   sinks=[redisSink],
                                   feature_names=['f_modified_streaming_count']
                                   )
    client.materialize_features(settings)  # Will streaming for 10 seconds since streamingTimeoutMs is 10000

[hangfei]

Is schema required? can it work without schema like frame-offline?

[jaymo001]

Yes. I believe it is. Kafka returns a bytes array of any type, so we have to rely on the schema to create a decoder.

[xiaoyongzhu]

Another bug while testing: We should make this kafka thing optional if the config does not have a kafka section, it will fail:

    self.kafka_endpoint = envutils.get_environment_variable_with_default(
  File "c:\users\xiaoyzhu\documents\github\feathr\feathr_project\feathr\_envvariableutil.py", line 43, in get_environment_variable_with_default
    yaml_layer = yaml_layer[arg]
KeyError: 'kafka'

[hangfei]

Discussed offline. Ideally we want users to define features without schema as well so it's aligned with offline and online. But seems Kafka doesn't provide such capability by itself yet. So instead, we will ask users to provide a schema.

In the future, if there is interface to allow kafka read without schema, we can make the schema optional.

[hangfei]

Could you add a dev guide as well for future devs?

[hangfei]

How do we support other streaming sources, like Flink?

Maybe we should do:

stream_source = StreamingSource(name="kafkaStreamingSource",
                              kafkaConfig=KafkaConfig(brokers=["feathrazureci.servicebus.windows.net:9093"],
                                                      topics=["feathrcieventhub"],
                                                      schema=schema)
                              )

[hangfei]

How do we support other streaming sources, like Flink?

Maybe we should do:

stream_source = StreamingSource(name="kafkaStreamingSource",
                              kafkaConfig=KafkaConfig(brokers=["feathrazureci.servicebus.windows.net:9093"],
                                                      topics=["feathrcieventhub"],
                                                      schema=schema)
                              )

Discussed offline. The current syntax just work so no need to introduce anohter layer.