|
37 | 37 | from apache_beam.options.value_provider import StaticValueProvider |
38 | 38 | from apache_beam.options.value_provider import ValueProvider |
39 | 39 | from apache_beam.transforms.display import HasDisplayData |
| 40 | +from apache_beam.utils import logs |
40 | 41 | from apache_beam.utils import proto_utils |
41 | 42 |
|
42 | 43 | __all__ = [ |
@@ -422,9 +423,10 @@ def get_all_options( |
422 | 423 | known_args, unknown_args = parser.parse_known_args(self._flags) |
423 | 424 | if retain_unknown_options: |
424 | 425 | if unknown_args: |
425 | | - _LOGGER.warning( |
426 | | - 'Unknown pipeline options received: %s. Ignore if flags are ' |
427 | | - 'used for internal purposes.' % (','.join(unknown_args))) |
| 426 | + if logs.allow_log_once(): |
| 427 | + _LOGGER.warning( |
| 428 | + 'Unknown pipeline options received: %s. Ignore if flags are ' |
| 429 | + 'used for internal purposes.' % (','.join(unknown_args))) |
428 | 430 |
|
429 | 431 | seen = set() |
430 | 432 |
|
@@ -466,7 +468,8 @@ def add_new_arg(arg, **kwargs): |
466 | 468 | parsed_args, _ = parser.parse_known_args(self._flags) |
467 | 469 | else: |
468 | 470 | if unknown_args: |
469 | | - _LOGGER.warning("Discarding unparseable args: %s", unknown_args) |
| 471 | + if logs.allow_log_once(): |
| 472 | + _LOGGER.warning("Discarding unparseable args: %s", unknown_args) |
470 | 473 | parsed_args = known_args |
471 | 474 | result = vars(parsed_args) |
472 | 475 |
|
@@ -1064,19 +1067,21 @@ def _warn_if_soft_delete_policy_enabled(self, arg_name): |
1064 | 1067 | return |
1065 | 1068 |
|
1066 | 1069 | gcs_path = getattr(self, arg_name, None) |
1067 | | - try: |
1068 | | - from apache_beam.io.gcp import gcsio |
1069 | | - if gcsio.GcsIO().is_soft_delete_enabled(gcs_path): |
| 1070 | + if logs.allow_log_once(gcs_path): |
| 1071 | + try: |
| 1072 | + from apache_beam.io.gcp import gcsio |
| 1073 | + if gcsio.GcsIO().is_soft_delete_enabled(gcs_path): |
| 1074 | + _LOGGER.warning( |
| 1075 | + "Bucket specified in %s has soft-delete policy enabled." |
| 1076 | + " To avoid being billed for unnecessary storage costs, turn" |
| 1077 | + " off the soft delete feature on buckets that your Dataflow" |
| 1078 | + " jobs use for temporary and staging storage. For more" |
| 1079 | + " information, see" |
| 1080 | + " https://cloud.google.com/storage/docs/use-soft-delete" |
| 1081 | + "#remove-soft-delete-policy." % arg_name) |
| 1082 | + except ImportError: |
1070 | 1083 | _LOGGER.warning( |
1071 | | - "Bucket specified in %s has soft-delete policy enabled." |
1072 | | - " To avoid being billed for unnecessary storage costs, turn" |
1073 | | - " off the soft delete feature on buckets that your Dataflow" |
1074 | | - " jobs use for temporary and staging storage. For more" |
1075 | | - " information, see" |
1076 | | - " https://cloud.google.com/storage/docs/use-soft-delete" |
1077 | | - "#remove-soft-delete-policy." % arg_name) |
1078 | | - except ImportError: |
1079 | | - _LOGGER.warning('Unable to check soft delete policy due to import error.') |
| 1084 | + 'Unable to check soft delete policy due to import error.') |
1080 | 1085 |
|
1081 | 1086 | # If either temp or staging location has an issue, we use the valid one for |
1082 | 1087 | # both locations. If both are bad we return an error. |
|
0 commit comments