Re: how to enable debugging mode for python worker harness

[Lydian Lee]

Hi XQ,

Sorry to bother you again, but I've tested the same thing again in a linux
env, and it is still not working and showing the same error in the python
worker harness.  (Note that this won't fail immediately, but it is failing
after the task is assigned to task manager and the python worker harness is
starting to work)

Wondering if you can share what you've changed  (maybe a PR) so that I can
test again on my linux machine. Thanks so much for your help.  There's
someone else also pinging me on the same error when testing, and I do want
to make this work for everyone.   Thanks!



On Mon, Mar 18, 2024 at 6:24 PM XQ Hu via user  wrote:

> I did not do anything special but ran `docker-compose -f
> docker-compose.yaml up` from your repo.
>
> On Sun, Mar 17, 2024 at 11:38 PM Lydian Lee 
> wrote:
>
>> Hi XQ,
>>
>> The code is simplified from my previous work and thus it is still using
>> the old version. But I've tested with Beam 2.54.0 and the code still works
>> (I mean using my company's image.)  If this is running well in your linux,
>> I guess there could be something related to how I build the docker image.
>> Curious if you could share the image you built to docker.io so that I
>> can confirm if the problem is related to only the image, thanks.
>>
>> The goal for this repo is to complete my previous talk:
>> https://www.youtube.com/watch?v=XUz90LpGAgc_channel=ApacheBeam
>>
>> On Sun, Mar 17, 2024 at 8:07 AM XQ Hu via user 
>> wrote:
>>
>>> I cloned your repo on my Linux machine, which is super useful to run.
>>> Not sure why you use Beam 2.41 but anyway, I tried this on my Linux machine:
>>>
>>> python t.py \
>>>   --topic test --group test-group --bootstrap-server localhost:9092 \
>>>   --job_endpoint localhost:8099 \
>>>   --artifact_endpoint localhost:8098 \
>>>   --environment_type=EXTERNAL \
>>>   --environment_config=localhost:5
>>>
>>> Note I replaced host.docker.internal with localhost and it runs well.
>>>
>>> I then tried to use host.docker.internal and it also runs well,
>>>
>>> Maybe this is related to your Mac setting?
>>>
>>> On Sun, Mar 17, 2024 at 8:34 AM Lydian Lee 
>>> wrote:
>>>
>>>>
>>>> Hi,
>>>>
>>>> Just FYI, the similar things works on a different image with the one I
>>>> built using my company’s image as base image. I’ve only replaced the base
>>>> image with ubuntu. But given that the error log is completely not helpful,
>>>> it’s really hard for me to continue debugging on the issue though.
>>>>
>>>> The docker is not required on my base image as I’ve already add extra
>>>> args to ReadFromKafka with default environment to be Process. This is proof
>>>> to work with my company’s docker image. For the host.internal.docker which
>>>> is also supported by docker for mac. The only thing i need to do is to
>>>> configure /etc/hosts so that i can submit the job directly from the laptop
>>>> and not the flink master.
>>>>
>>>>
>>>>
>>>>
>>>> On Sun, Mar 17, 2024 at 2:40 AM Jaehyeon Kim  wrote:
>>>>
>>>>> Hello,
>>>>>
>>>>> The pipeline runs in host while host.docker.internal would only be
>>>>> resolved on the containers that run with the host network mode. I guess 
>>>>> the
>>>>> pipeline wouldn't be accessible to host.docker.internal and fails to run.
>>>>>
>>>>> If everything before ReadFromKafka works successfully, a docker
>>>>> container will be launched with the host network mode so that
>>>>> host.docker.internal:9092 can be resolved inside the container. As far as
>>>>> I've checked, however, it fails when I start a flink cluster on docker and
>>>>> I had to rely on a local flink cluster. If you'd like to try to use 
>>>>> docker,
>>>>> you should have docker installed on your custom docker image and
>>>>> volume-map /var/run/docker.sock to the flink task manager. Otherwise, it
>>>>> won't be able to launch a Docker container for reading kafka messages.
>>>>>
>>>>> Cheers,
>>>>> Jaehyeon
>>>>>
>>>>>
>>>>> On Sun, 17 Mar 2024 at 18:21, Lydian Lee 
>>>>> wrote:
>>>>>
>>>>>> Hi,
>>>>>>
>>>>>> I have an issue when setting up a POC of  Python SDK with Fli

Creating custom metrics with labels (python SDK + Flink Runner)

[Lydian Lee]

Hi,

I am using beam python SDK with flink runner, and I am trying to add custom
labels to the metrics.

It seems like the provided function (link
)
doesn't allow me to add labels.

```
 @staticmethod
  def counter(namespace, name):
```

Taking a deeper look on the code, it is using `MetricName`:
```
return Metrics.DelegatingCounter(MetricName(namespace, name))
```

and the MetricName does support for labels (link

):
```
 def __init__(self, namespace, name, urn=None, labels=None):
```


Therefore I am trying to manually update the metricName labels by creating
a new class that does pass the labels:
```
class MetricWithLabels(object):

@staticmethod
def counter(namespace, name, labels):
namespace = beam.metrics.Metrics.get_namespace(namespace)
return
beam.metrics.Metrics.DelegatingCounter(beam.metrics.metricbase.MetricName(namespace,
name, labels=labels))
```

However, when I test this in code
```
class addExample(beam.DoFn):

def __init__(self):
beam.DoFn.__init__(self)
self.example_counter = MetricWithLabels.counter('example_counter',
{'label1': 'value1'})


def process(self, element):
self.example_counter.inc()
return element
```

The final output still don't have the labels I created, and it seems like
the output metrics is having labels completely overwrite by Flink runner.:
```
flink_taskmanager_job_task_operator_custom_example_counter{job_id="41b6a8a793e47f8edaa91e2dccc90a5f",task_id="ecddd3e2ade3edda3cd8430d7b243742",task_attempt_id="72497124f6cc6d7c73674fcbba0664b4",host="172.19.0.5",operator_id="04ff9a876e7d6fb91800bbfb5f72ce25",operator_name="[3]{Convert
format, metrics example, logging}",task_name="Source: Reading message from
kafka/Read from kafka topic
['test']/KafkaIO.Read/KafkaIO.Read.ReadFromKafkaViaUnbounded/Read(KafkaUnboundedSource)
-> Flat Map -> Map -> [1]Reading message from kafka/Read from kafka topic
['test']/Remove Kafka Metadata -> [3]{Convert format, metrics example,
logging}",task_attempt_num="0",job_name="None",tm_id="172.19.0.5:37847
-bb580c",subtask_index="0",}
```

I wonder if there's a way that we can pass in the custom labels to the
metrics with flink runner, and if not, where are the labels being
overwritten so that I might be able to update the code to support custom
labels. Thanks

Re: how to enable debugging mode for python worker harness

[Lydian Lee]

Hi XQ,

The code is simplified from my previous work and thus it is still using the
old version. But I've tested with Beam 2.54.0 and the code still works (I
mean using my company's image.)  If this is running well in your linux, I
guess there could be something related to how I build the docker image.
Curious if you could share the image you built to docker.io so that I can
confirm if the problem is related to only the image, thanks.

The goal for this repo is to complete my previous talk:
https://www.youtube.com/watch?v=XUz90LpGAgc_channel=ApacheBeam

On Sun, Mar 17, 2024 at 8:07 AM XQ Hu via user  wrote:

> I cloned your repo on my Linux machine, which is super useful to run. Not
> sure why you use Beam 2.41 but anyway, I tried this on my Linux machine:
>
> python t.py \
>   --topic test --group test-group --bootstrap-server localhost:9092 \
>   --job_endpoint localhost:8099 \
>   --artifact_endpoint localhost:8098 \
>   --environment_type=EXTERNAL \
>   --environment_config=localhost:5
>
> Note I replaced host.docker.internal with localhost and it runs well.
>
> I then tried to use host.docker.internal and it also runs well,
>
> Maybe this is related to your Mac setting?
>
> On Sun, Mar 17, 2024 at 8:34 AM Lydian Lee 
> wrote:
>
>>
>> Hi,
>>
>> Just FYI, the similar things works on a different image with the one I
>> built using my company’s image as base image. I’ve only replaced the base
>> image with ubuntu. But given that the error log is completely not helpful,
>> it’s really hard for me to continue debugging on the issue though.
>>
>> The docker is not required on my base image as I’ve already add extra
>> args to ReadFromKafka with default environment to be Process. This is proof
>> to work with my company’s docker image. For the host.internal.docker which
>> is also supported by docker for mac. The only thing i need to do is to
>> configure /etc/hosts so that i can submit the job directly from the laptop
>> and not the flink master.
>>
>>
>>
>>
>> On Sun, Mar 17, 2024 at 2:40 AM Jaehyeon Kim  wrote:
>>
>>> Hello,
>>>
>>> The pipeline runs in host while host.docker.internal would only be
>>> resolved on the containers that run with the host network mode. I guess the
>>> pipeline wouldn't be accessible to host.docker.internal and fails to run.
>>>
>>> If everything before ReadFromKafka works successfully, a docker
>>> container will be launched with the host network mode so that
>>> host.docker.internal:9092 can be resolved inside the container. As far as
>>> I've checked, however, it fails when I start a flink cluster on docker and
>>> I had to rely on a local flink cluster. If you'd like to try to use docker,
>>> you should have docker installed on your custom docker image and
>>> volume-map /var/run/docker.sock to the flink task manager. Otherwise, it
>>> won't be able to launch a Docker container for reading kafka messages.
>>>
>>> Cheers,
>>> Jaehyeon
>>>
>>>
>>> On Sun, 17 Mar 2024 at 18:21, Lydian Lee 
>>> wrote:
>>>
>>>> Hi,
>>>>
>>>> I have an issue when setting up a POC of  Python SDK with Flink runner
>>>> to run in docker-compose.  The python worker harness was not returning any
>>>> error but:
>>>> ```
>>>> python-worker-harness-1  | 2024/03/17 07:10:17 Executing: python -m
>>>> apache_beam.runners.worker.sdk_worker_main
>>>> python-worker-harness-1  | 2024/03/17 07:10:24 Python exited: 
>>>> ```
>>>> and dead.  The error message seems totally unuseful, and I am wondering
>>>> if there's a way to make the harness script show more debug logging.
>>>>
>>>> I started my harness via:
>>>> ```
>>>> /opt/apache/beam/boot --worker_pool
>>>> ```
>>>> and configure my script to use the harness
>>>> ```
>>>> python docker/src/example.py \
>>>>   --topic test --group test-group --bootstrap-server
>>>> host.docker.internal:9092 \
>>>>   --job_endpoint host.docker.internal:8099 \
>>>>   --artifact_endpoint host.docker.internal:8098 \
>>>>   --environment_type=EXTERNAL \
>>>>   --environment_config=host.docker.internal:5
>>>> ```
>>>> The full settings is available in:
>>>> https://github.com/lydian/beam-python-flink-runner-examples
>>>> Thanks for your help
>>>>
>>>>

Re: how to enable debugging mode for python worker harness

[Lydian Lee]

Hi,

Just FYI, the similar things works on a different image with the one I
built using my company’s image as base image. I’ve only replaced the base
image with ubuntu. But given that the error log is completely not helpful,
it’s really hard for me to continue debugging on the issue though.

The docker is not required on my base image as I’ve already add extra args
to ReadFromKafka with default environment to be Process. This is proof to
work with my company’s docker image. For the host.internal.docker which is
also supported by docker for mac. The only thing i need to do is to
configure /etc/hosts so that i can submit the job directly from the laptop
and not the flink master.




On Sun, Mar 17, 2024 at 2:40 AM Jaehyeon Kim  wrote:

> Hello,
>
> The pipeline runs in host while host.docker.internal would only be
> resolved on the containers that run with the host network mode. I guess the
> pipeline wouldn't be accessible to host.docker.internal and fails to run.
>
> If everything before ReadFromKafka works successfully, a docker container
> will be launched with the host network mode so that
> host.docker.internal:9092 can be resolved inside the container. As far as
> I've checked, however, it fails when I start a flink cluster on docker and
> I had to rely on a local flink cluster. If you'd like to try to use docker,
> you should have docker installed on your custom docker image and
> volume-map /var/run/docker.sock to the flink task manager. Otherwise, it
> won't be able to launch a Docker container for reading kafka messages.
>
> Cheers,
> Jaehyeon
>
>
> On Sun, 17 Mar 2024 at 18:21, Lydian Lee  wrote:
>
>> Hi,
>>
>> I have an issue when setting up a POC of  Python SDK with Flink runner to
>> run in docker-compose.  The python worker harness was not returning any
>> error but:
>> ```
>> python-worker-harness-1  | 2024/03/17 07:10:17 Executing: python -m
>> apache_beam.runners.worker.sdk_worker_main
>> python-worker-harness-1  | 2024/03/17 07:10:24 Python exited: 
>> ```
>> and dead.  The error message seems totally unuseful, and I am wondering
>> if there's a way to make the harness script show more debug logging.
>>
>> I started my harness via:
>> ```
>> /opt/apache/beam/boot --worker_pool
>> ```
>> and configure my script to use the harness
>> ```
>> python docker/src/example.py \
>>   --topic test --group test-group --bootstrap-server
>> host.docker.internal:9092 \
>>   --job_endpoint host.docker.internal:8099 \
>>   --artifact_endpoint host.docker.internal:8098 \
>>   --environment_type=EXTERNAL \
>>   --environment_config=host.docker.internal:5
>> ```
>> The full settings is available in:
>> https://github.com/lydian/beam-python-flink-runner-examples
>> Thanks for your help
>>
>>

how to enable debugging mode for python worker harness

[Lydian Lee]

Hi,

I have an issue when setting up a POC of  Python SDK with Flink runner to
run in docker-compose.  The python worker harness was not returning any
error but:
```
python-worker-harness-1  | 2024/03/17 07:10:17 Executing: python -m
apache_beam.runners.worker.sdk_worker_main
python-worker-harness-1  | 2024/03/17 07:10:24 Python exited: 
```
and dead.  The error message seems totally unuseful, and I am wondering if
there's a way to make the harness script show more debug logging.

I started my harness via:
```
/opt/apache/beam/boot --worker_pool
```
and configure my script to use the harness
```
python docker/src/example.py \
  --topic test --group test-group --bootstrap-server
host.docker.internal:9092 \
  --job_endpoint host.docker.internal:8099 \
  --artifact_endpoint host.docker.internal:8098 \
  --environment_type=EXTERNAL \
  --environment_config=host.docker.internal:5
```
The full settings is available in:
https://github.com/lydian/beam-python-flink-runner-examples
Thanks for your help

Request to join slack channel

[Lydian Lee]

Hi,

Can I get the invitation to join slack channel ?  The ASF slack seems
required invitation to be able to join. Thanks