<commit_before># Copyright 2016 Google Inc. All Rights Reserved.
#
# 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.

"""Python Dataflow streaming worker."""

from __future__ import absolute_import

import logging
import random
import time
import traceback


from grpc.beta import implementations

from google.cloud.dataflow.internal import windmill_pb2
from google.cloud.dataflow.internal import windmill_service_pb2
from google.cloud.dataflow.utils import retry
from google.cloud.dataflow.worker import executor
from google.cloud.dataflow.worker import maptask
from google.cloud.dataflow.worker import windmillio
from google.cloud.dataflow.worker import windmillstate
import apitools.base.py as apitools_base
import apitools.clients.dataflow as dataflow


# pylint: disable=invalid-name
class WindmillClient(object):
  """Client for communication with Windmill."""

  def __init__(self, host, port, request_timeout=10):
    self.host = host
    self.port = port
    self.request_timeout = request_timeout

    channel = implementations.insecure_channel(host, port)
    self.stub = (
        windmill_service_pb2.beta_create_CloudWindmillServiceV1Alpha1_stub(
            channel))

  @retry.with_exponential_backoff()
  def GetWork(self, request):
    return self.stub.GetWork(request, self.request_timeout)

  @retry.with_exponential_backoff()
  def GetData(self, request):
    return self.stub.GetData(request, self.request_timeout)

  @retry.with_exponential_backoff()
  def CommitWork(self, request):
    return self.stub.CommitWork(request, self.request_timeout)

  @retry.with_exponential_backoff()
  def GetConfig(self, request):
    return self.stub.GetConfig(request, self.request_timeout)

  @retry.with_exponential_backoff()
  def ReportStats(self, request):
    return self.stub.ReportStats(request, self.request_timeout)
# pylint: enable=invalid-name


class StreamingWorker(object):
  """A streaming worker that communicates with Windmill."""

  # Maximum size of the result of a GetWork request.
  MAX_GET_WORK_FETCH_BYTES = 64 << 20  # 64m
  # Maximum number of items to return in a GetWork request.
  MAX_GET_WORK_ITEMS = 100

  # TODO(altay): Remove windmill default port and host.
  WINDMILL_DEFAULT_PORT = 12355

  def __init__(self, properties):
    self.project_id = properties['project_id']
    self.job_id = properties['job_id']
    self.worker_id = properties['worker_id']

    self.client_id = random.getrandbits(63)
    windmill_host = properties.get('windmill.host', self.worker_id)
    windmill_port = int(properties.get('windmill.hostport',
                                       StreamingWorker.WINDMILL_DEFAULT_PORT))
    logging.info('Using gRPC to connect to Windmill at %s:%d.', windmill_host,
                 windmill_port)
    self.windmill = WindmillClient(windmill_host, windmill_port)<commit_msg>Remove windmill host and port defaults from streaming worker<commit_after># Copyright 2016 Google Inc. All Rights Reserved.
#
# 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.

"""Python Dataflow streaming worker."""

from __future__ import absolute_import

import logging
import random
import time
import traceback


from grpc.beta import implementations

from google.cloud.dataflow.internal import windmill_pb2
from google.cloud.dataflow.internal import windmill_service_pb2
from google.cloud.dataflow.utils import retry
from google.cloud.dataflow.worker import executor
from google.cloud.dataflow.worker import maptask
from google.cloud.dataflow.worker import windmillio
from google.cloud.dataflow.worker import windmillstate
import apitools.base.py as apitools_base
import apitools.clients.dataflow as dataflow


# pylint: disable=invalid-name
class WindmillClient(object):
  """Client for communication with Windmill."""

  def __init__(self, host, port, request_timeout=10):
    self.host = host
    self.port = port
    self.request_timeout = request_timeout

    channel = implementations.insecure_channel(host, port)
    self.stub = (
        windmill_service_pb2.beta_create_CloudWindmillServiceV1Alpha1_stub(
            channel))

  @retry.with_exponential_backoff()
  def GetWork(self, request):
    return self.stub.GetWork(request, self.request_timeout)

  @retry.with_exponential_backoff()
  def GetData(self, request):
    return self.stub.GetData(request, self.request_timeout)

  @retry.with_exponential_backoff()
  def CommitWork(self, request):
    return self.stub.CommitWork(request, self.request_timeout)

  @retry.with_exponential_backoff()
  def GetConfig(self, request):
    return self.stub.GetConfig(request, self.request_timeout)

  @retry.with_exponential_backoff()
  def ReportStats(self, request):
    return self.stub.ReportStats(request, self.request_timeout)
# pylint: enable=invalid-name


class StreamingWorker(object):
  """A streaming worker that communicates with Windmill."""

  # Maximum size of the result of a GetWork request.
  MAX_GET_WORK_FETCH_BYTES = 64 << 20  # 64m
  # Maximum number of items to return in a GetWork request.
  MAX_GET_WORK_ITEMS = 100

  def __init__(self, properties):
    self.project_id = properties['project_id']
    self.job_id = properties['job_id']
    self.worker_id = properties['worker_id']

    self.client_id = random.getrandbits(63)
    windmill_host = properties['windmill.host']
    windmill_port = int(properties['windmill.grpc_port'])
    logging.info('Using gRPC to connect to Windmill at %s:%d.', windmill_host,
                 windmill_port)
    self.windmill = WindmillClient(windmill_host, windmill_port)
{'fraction_non_alphanumeric': 0.08275862068965517, 'fraction_numerical': 0.008735632183908045, 'mean_word_length': 4.544604927782498, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7658027', 'n_tokens_mistral': 1940, 'n_tokens_neox': 1881, 'n_words': 552}