You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2020/02/12 19:00:40 UTC
[GitHub] [airflow] dferguson992 opened a new pull request #7407: Add
KafkaConsumerHook, KafkaProduerHook and KafkaSensor
dferguson992 opened a new pull request #7407: Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407
Dear Airflow Maintainers,
Please accept the following PR that
Add the KafkaProducerHook.
Add the KafkaConsumerHook.
Add the KafkaSensor which listens to messages with a specific topic.
Related Issue:
#1311
Reminder to contributors:
You must add an Apache License header to all new files
Please squash your commits when possible and follow the 7 rules of good Git commits
I am new to the community, I am not sure the files are at the right place or missing anything.
The sensor could be used as the first node of a dag where the second node can be a TriggerDagRunOperator. The messages are polled in a batch and the dag runs are dynamically generated.
Thanks!
Note, as per denied PR #1415, it is important to mention these integrations are not suitable for low-latency/high-throughput/streaming. For reference, https://github.com/apache/airflow/pull/1415#issuecomment-484429806.
Co-authored-by: Dan Ferguson dferguson992@gmail.com
Co-authored-by: Hanan Shteingart
Co-authored-by: YuanfΞi Zhu
---
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/master/UPDATING.md).
Read the [Pull Request Guidelines](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#pull-request-guidelines) for more information.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] kaxil commented on a change in pull request #7407:
[AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #7407: [AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407#discussion_r383521743
##########
File path: airflow/contrib/hooks/kafka_consumer_hook.py
##########
@@ -0,0 +1,71 @@
+# 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.
+
+from airflow.hooks import base_hook as BaseHook
+from kafka import KafkaConsumer
+
+
+class KafkaConsumerHook(BaseHook):
+
+ DEFAULT_HOST = 'localhost'
+ DEFAULT_PORT = 9092
+
+ def __init__(self, conn_id, topic):
+ super(KafkaConsumerHook, self).__init__(None)
+ self.conn = self.get_connection(conn_id)
Review comment:
Let's move this call to get_connection, example: https://github.com/apache/airflow/commit/4bec1cc489f5d19daf7450c75c3e8057c9709dbd
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] ashb commented on issue #7407: Add KafkaConsumerHook,
KafkaProduerHook and KafkaSensor
Posted by GitBox <gi...@apache.org>.
ashb commented on issue #7407: Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407#issuecomment-585395982
@dferguson992 This PR pre-dates the projects use of Jira so you'll need to sign up for an account at issues.apache.org/jira/ and create one, please.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] dmmakita commented on issue #7407: [AIRFLOW-6786] Add
KafkaConsumerHook, KafkaProduerHook and KafkaSensor
Posted by GitBox <gi...@apache.org>.
dmmakita commented on issue #7407: [AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407#issuecomment-613182747
Thank you for your response. I will try to do something.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] kaxil commented on a change in pull request #7407:
[AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #7407: [AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407#discussion_r383521933
##########
File path: airflow/contrib/hooks/kafka_consumer_hook.py
##########
@@ -0,0 +1,71 @@
+# 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.
+
+from airflow.hooks import base_hook as BaseHook
+from kafka import KafkaConsumer
+
+
+class KafkaConsumerHook(BaseHook):
+
+ DEFAULT_HOST = 'localhost'
+ DEFAULT_PORT = 9092
+
+ def __init__(self, conn_id, topic):
+ super(KafkaConsumerHook, self).__init__(None)
+ self.conn = self.get_connection(conn_id)
Review comment:
This is to avoid DB call on Initialisation.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] ashb commented on a change in pull request #7407:
[AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #7407: [AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407#discussion_r378509192
##########
File path: airflow/contrib/sensors/kafka_sensor.py
##########
@@ -0,0 +1,50 @@
+# 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.
+
+import logging
+
+from src.kafka_hook import KafkaConsumerHook
+from airflow.operators.sensors import BaseSensorOperator
+from airflow.utils import apply_defaults
+
+
+class KafkaSensor(BaseSensorOperator):
+ """
+ Consumes the Kafka message with the specific topic
+ """
+
+ @apply_defaults
+ def __init__(self, conn_id, topic, *args, **kwargs):
+ """
+ Initialize the sensor, the connection establish
+ is put off to it's first time usage.
+ :param conn_id:
+ the kafka broker connection whom this sensor
+ subscripts against.
+ :param topic:
+ the subscribed topic
+ """
+ self.topic = topic
+ self.hook = KafkaConsumerHook(conn_id, topic)
Review comment:
Please don't create hooks in init.
Instead store conn_id and topic on self and then do
```
@cached_property
def hook(self):
return KafkaConsumerHook(self.conn_id, self.topic)
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] ashb commented on issue #7407: [AIRFLOW-6786] Add
KafkaConsumerHook, KafkaProduerHook and KafkaSensor
Posted by GitBox <gi...@apache.org>.
ashb commented on issue #7407: [AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407#issuecomment-589729615
@dferguson992 I've force pushed this branch to update the format/add the emails.
I couldn't see where you got "Hanan Shteingart" from as an author- which PR was that?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] ashb commented on a change in pull request #7407:
[AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #7407: [AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407#discussion_r378507748
##########
File path: airflow/contrib/hooks/kafka_consumer_hook.py
##########
@@ -0,0 +1,69 @@
+# 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.
+
+from airflow.hooks import base_hook as BaseHook
+from kafka import KafkaConsumer
+
+
+class KafkaConsumerHook(BaseHook):
+
+ default_host = 'localhost'
Review comment:
`DEFAULT_HOST` etc for constants please
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] ashb commented on issue #7407: [AIRFLOW-6786] Add
KafkaConsumerHook, KafkaProduerHook and KafkaSensor
Posted by GitBox <gi...@apache.org>.
ashb commented on issue #7407: [AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407#issuecomment-585413257
The first thing I notice is that we now need to rename this:
- airflow/contrib/hooks/kafka_consumer_hook.py
->
- airflow/providers/apache/kafka/hooks/kafka_consumer.py
etc. A bit more detail is in https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-21%3A+Changes+in+import+paths, shout if that's not clear.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] dmmakita commented on issue #7407: [AIRFLOW-6786] Add
KafkaConsumerHook, KafkaProduerHook and KafkaSensor
Posted by GitBox <gi...@apache.org>.
dmmakita commented on issue #7407: [AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407#issuecomment-612543866
Hi, @dferguson992, I'm having similar needs related to airflow-kafka. May I try to help you with this issue #7407?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] dferguson992 commented on issue #7407: [AIRFLOW-6786] Add
KafkaConsumerHook, KafkaProduerHook and KafkaSensor
Posted by GitBox <gi...@apache.org>.
dferguson992 commented on issue #7407: [AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407#issuecomment-597912749
hey @ashb I'm not sure what else to do for this PR. Is there anything outstanding before it can be merged? I fixed as many of the travis CI issues as I could not sure what else to do for this
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] dferguson992 commented on issue #7407: [AIRFLOW-6786] Add
KafkaConsumerHook, KafkaProduerHook and KafkaSensor
Posted by GitBox <gi...@apache.org>.
dferguson992 commented on issue #7407: [AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407#issuecomment-606616677
Hey @ashb hope all is well. I'm not sure what to do to get the Travis CI checks to pass, but the code is functional and includes your suggestions. Can we move forward with the PR?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] dferguson992 commented on issue #7407: [AIRFLOW-6786] Add
KafkaConsumerHook, KafkaProduerHook and KafkaSensor
Posted by GitBox <gi...@apache.org>.
dferguson992 commented on issue #7407: [AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407#issuecomment-598728913
@ashb I got that co-author by mistake. on the original PR, that user was the last to comment on it. but i mistakenly took that user as the PR author, not an interested third-party as I should have. I've removed that user from the PR notes and the corresponding issue.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] dferguson992 commented on a change in pull request #7407:
[AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
Posted by GitBox <gi...@apache.org>.
dferguson992 commented on a change in pull request #7407: [AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407#discussion_r386641018
##########
File path: airflow/contrib/hooks/kafka_consumer_hook.py
##########
@@ -0,0 +1,71 @@
+# 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.
+
+from airflow.hooks import base_hook as BaseHook
+from kafka import KafkaConsumer
+
+
+class KafkaConsumerHook(BaseHook):
+
+ DEFAULT_HOST = 'localhost'
+ DEFAULT_PORT = 9092
+
+ def __init__(self, conn_id, topic):
+ super(KafkaConsumerHook, self).__init__(None)
+ self.conn = self.get_connection(conn_id)
Review comment:
Address in newest commit.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] ashb commented on issue #7407: [AIRFLOW-6786] Add
KafkaConsumerHook, KafkaProduerHook and KafkaSensor
Posted by GitBox <gi...@apache.org>.
ashb commented on issue #7407: [AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407#issuecomment-585406957
@dferguson992 Good work!. I'll take a look in detail tomorrow. (there will likely be a few rounds of review as this code was quite old)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] ashb commented on a change in pull request #7407:
[AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #7407: [AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407#discussion_r378508501
##########
File path: airflow/contrib/hooks/kafka_producer_hook.py
##########
@@ -0,0 +1,56 @@
+# 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.
+
+from airflow.hooks import base_hook as BaseHook
+from kafka import KafkaProducer
+
+
+class KafkaProducerHook(BaseHook):
+
+ default_host = 'localhost'
+ default_port = 9092
+
+ def __init__(self, conn_id, topic):
+ super(KafkaConsumerHook, self).__init__(None)
+ self.conn = self.get_connection(conn_id)
+ self.server = None
+ self.consumer = None
+ self.producer = None
+ self.topic = topic
+
+ def get_conn(self):
+ conf = self.conn.extra_dejson
+ host = self.conn.host or self.default_host
+ port = self.conn.port or self.default_port
+
+ conf['enable_auto_commit'] = False
+ self.server = '{host}:{port}'.format(**locals())
+ self.producer = KafkaProducer(
+ bootstrap_servers=self.server, **conf)
+
+ return self.producer
+
+ def send_message(self, topic, value=None, key=None, headers=None, partition=None, timestamp_ms=None):
+ producer = self.get_pub()
+ future_record_metadata = producer.send(topic, value=value, key=key, headers=headers, partition=partition,
+ timestamp_ms=timestamp_ms)
+ producer.close()
Review comment:
This close should probably be in a `finally` block.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] dferguson992 commented on issue #7407: [AIRFLOW-6786] Add
KafkaConsumerHook, KafkaProduerHook and KafkaSensor
Posted by GitBox <gi...@apache.org>.
dferguson992 commented on issue #7407: [AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407#issuecomment-587479117
@ashb I've made the recommended changes except for the github email address. How does one acquire someone else's GH email?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] ashb commented on a change in pull request #7407:
[AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #7407: [AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407#discussion_r378507632
##########
File path: airflow/contrib/hooks/kafka_consumer_hook.py
##########
@@ -0,0 +1,69 @@
+# 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.
+
+from airflow.hooks import base_hook as BaseHook
+from kafka import KafkaConsumer
+
+
+class KafkaConsumerHook(BaseHook):
+
+ default_host = 'localhost'
+ default_port = 9092
+
+ def __init__(self, conn_id, topic):
+ super(KafkaConsumerHook, self).__init__(None)
+ self.conn = self.get_connection(conn_id)
+ self.server = None
+ self.consumer = None
+ self.producer = None
+ self.topic = topic
+
+ def get_conn(self):
+ conf = self.conn.extra_dejson
+ host = self.conn.host or self.default_host
+ port = self.conn.port or self.default_port
+
+ # Disable auto commit as the hook will commit right
+ # after polling.
+ conf['enable_auto_commit'] = False
+
+ self.server = '{host}:{port}'.format(**locals())
+ self.consumer = KafkaConsumer(
+ self.topic,
+ bootstrap_servers=self.server, **conf)
+
+ return self.consumer
+
+ def get_messages(self):
+ """
+ Get all the messages haven't been consumed, it doesn't
+ block by default, then commit the offset.
+ :return:
+ A list of messages
+ """
+ consumer = self.get_conn()
+
+ # `poll` returns a dict where keys are the partitions
+ # and values are the corresponding messages.
+ messages = consumer.poll(timeout_ms=50)
Review comment:
timeout should be a parameter (default of 50 is fine)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] boring-cyborg[bot] commented on issue #7407: Add
KafkaConsumerHook, KafkaProduerHook and KafkaSensor
Posted by GitBox <gi...@apache.org>.
boring-cyborg[bot] commented on issue #7407: Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407#issuecomment-585364531
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst)
Here are some useful points:
- Pay attention to the quality of your code (flake8, pylint and type annotations). Our [pre-commits]( https://github.com/apache/airflow/blob/master/STATIC_CODE_CHECKS.rst#prerequisites-for-pre-commit-hooks) will help you with that.
- In case of a new feature add useful documentation (in docstrings or in `docs/` directory). Adding a new operator? Check this short [guide](https://github.com/apache/airflow/blob/master/docs/howto/custom-operator.rst) Consider adding an example DAG that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/master/BREEZE.rst) for testing locally, it’s a heavy docker but it ships with a working Airflow and a lot of integrations.
- Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://apache-airflow-slack.herokuapp.com/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] dferguson992 commented on issue #7407: [AIRFLOW-6786] Add
KafkaConsumerHook, KafkaProduerHook and KafkaSensor
Posted by GitBox <gi...@apache.org>.
dferguson992 commented on issue #7407: [AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407#issuecomment-585410381
@ashb thanks! i'll be on vacation, back on Tuesday of next week. Happy to make any adjustments or additions to the code-base once I return. The code is quite old you're right, but it should be pretty forward compatible with vanilla Kafka.
I'm not sure if this project shies away from branding, but it may be worthwhile to consider adding support for Confluent Kafka specifically.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] ashb commented on a change in pull request #7407:
[AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #7407: [AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407#discussion_r378508163
##########
File path: airflow/contrib/hooks/kafka_producer_hook.py
##########
@@ -0,0 +1,56 @@
+# 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.
+
+from airflow.hooks import base_hook as BaseHook
+from kafka import KafkaProducer
+
+
+class KafkaProducerHook(BaseHook):
+
+ default_host = 'localhost'
+ default_port = 9092
+
+ def __init__(self, conn_id, topic):
+ super(KafkaConsumerHook, self).__init__(None)
+ self.conn = self.get_connection(conn_id)
+ self.server = None
+ self.consumer = None
+ self.producer = None
+ self.topic = topic
+
+ def get_conn(self):
+ conf = self.conn.extra_dejson
+ host = self.conn.host or self.default_host
+ port = self.conn.port or self.default_port
+
+ conf['enable_auto_commit'] = False
+ self.server = '{host}:{port}'.format(**locals())
Review comment:
Please avoid `**locals`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] ashb commented on issue #7407: [AIRFLOW-6786] Add
KafkaConsumerHook, KafkaProduerHook and KafkaSensor
Posted by GitBox <gi...@apache.org>.
ashb commented on issue #7407: [AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407#issuecomment-598710287
> @dferguson992 I've force pushed this branch to update the format/add the emails.
>
> I couldn't see where you got "Hanan Shteingart" from as an author- which PR was that?
@dferguson992 Can you answer this Q?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] ashb commented on a change in pull request #7407:
[AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #7407: [AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407#discussion_r378508326
##########
File path: airflow/contrib/hooks/kafka_producer_hook.py
##########
@@ -0,0 +1,56 @@
+# 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.
+
+from airflow.hooks import base_hook as BaseHook
+from kafka import KafkaProducer
+
+
+class KafkaProducerHook(BaseHook):
+
+ default_host = 'localhost'
+ default_port = 9092
+
+ def __init__(self, conn_id, topic):
+ super(KafkaConsumerHook, self).__init__(None)
+ self.conn = self.get_connection(conn_id)
+ self.server = None
+ self.consumer = None
+ self.producer = None
+ self.topic = topic
+
+ def get_conn(self):
+ conf = self.conn.extra_dejson
+ host = self.conn.host or self.default_host
+ port = self.conn.port or self.default_port
+
+ conf['enable_auto_commit'] = False
+ self.server = '{host}:{port}'.format(**locals())
+ self.producer = KafkaProducer(
+ bootstrap_servers=self.server, **conf)
+
+ return self.producer
+
+ def send_message(self, topic, value=None, key=None, headers=None, partition=None, timestamp_ms=None):
+ producer = self.get_pub()
Review comment:
I don't see a `get_pub` method?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] dferguson992 commented on issue #7407: [AIRFLOW-6786] Add
KafkaConsumerHook, KafkaProduerHook and KafkaSensor
Posted by GitBox <gi...@apache.org>.
dferguson992 commented on issue #7407: [AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407#issuecomment-612929452
> Hi, @dferguson992, I'm having similar needs related to airflow-kafka. May I try to help you with this issue #7407?
That would be awesome! But I'm not sure what else needs doing. If you know more on this, please do go for it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] ashb commented on a change in pull request #7407:
[AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #7407: [AIRFLOW-6786] Add KafkaConsumerHook, KafkaProduerHook and KafkaSensor
URL: https://github.com/apache/airflow/pull/7407#discussion_r378509406
##########
File path: airflow/contrib/sensors/kafka_sensor.py
##########
@@ -0,0 +1,50 @@
+# 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.
+
+import logging
+
+from src.kafka_hook import KafkaConsumerHook
+from airflow.operators.sensors import BaseSensorOperator
+from airflow.utils import apply_defaults
+
+
+class KafkaSensor(BaseSensorOperator):
+ """
+ Consumes the Kafka message with the specific topic
+ """
+
+ @apply_defaults
+ def __init__(self, conn_id, topic, *args, **kwargs):
+ """
+ Initialize the sensor, the connection establish
+ is put off to it's first time usage.
+ :param conn_id:
+ the kafka broker connection whom this sensor
+ subscripts against.
+ :param topic:
+ the subscribed topic
+ """
+ self.topic = topic
Review comment:
This would be nice to have templates, so please add this at the class level:
```
templated_fields = ( 'topic', )
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services