| 57 | NUM_LINES_OFFSET_TRACE = int(os.environ.get('PYTHON_TRACE_OFFSET', 0)) |
| 58 | |
| 59 | class PythonStreamingController: |
| 60 | def __init__(self, profiling_mode=False): |
| 61 | self.profiling_mode = profiling_mode |
| 62 | |
| 63 | self.input_count = 0 |
| 64 | self.next_input_count_to_log = 1 |
| 65 | |
| 66 | def main(self, |
| 67 | module_name, file_path, func_name, cache_path, |
| 68 | output_stream_path, error_stream_path, log_file_name, is_illustrate_str): |
| 69 | sys.stdin = os.fdopen(sys.stdin.fileno(), 'rb', 0) |
| 70 | |
| 71 | #Need to ensure that user functions can't write to the streams we use to |
| 72 | #communicate with pig. |
| 73 | self.stream_output = os.fdopen(sys.stdout.fileno(), 'wb', 0) |
| 74 | self.stream_error = os.fdopen(sys.stderr.fileno(), 'wb', 0) |
| 75 | |
| 76 | self.input_stream = sys.stdin |
| 77 | self.output_stream = open(output_stream_path, 'a') |
| 78 | sys.stderr = open(error_stream_path, 'w') |
| 79 | is_illustrate = is_illustrate_str == "true" |
| 80 | |
| 81 | sys.path.append(file_path) |
| 82 | sys.path.append(cache_path) |
| 83 | sys.path.append('.') |
| 84 | |
| 85 | logging.basicConfig(filename=log_file_name, format="%(asctime)s %(levelname)s %(message)s", level=udf_logging.udf_log_level) |
| 86 | logging.info("To reduce the amount of information being logged only a small subset of rows are logged at the INFO level. Call udf_logging.set_log_level_debug in pig_util to see all rows being processed.") |
| 87 | |
| 88 | input_str = self.get_next_input() |
| 89 | |
| 90 | try: |
| 91 | func = __import__(module_name, globals(), locals(), [func_name], -1).__dict__[func_name] |
| 92 | except: |
| 93 | #These errors should always be caused by user code. |
| 94 | write_user_exception(module_name, self.stream_error, NUM_LINES_OFFSET_TRACE) |
| 95 | self.close_controller(-1) |
| 96 | |
| 97 | if is_illustrate or udf_logging.udf_log_level != logging.DEBUG: |
| 98 | #Only log output for illustrate after we get the flag to capture output. |
| 99 | sys.stdout = open(os.devnull, 'w') |
| 100 | else: |
| 101 | sys.stdout = self.output_stream |
| 102 | |
| 103 | while input_str != END_OF_STREAM: |
| 104 | should_log = False |
| 105 | if self.input_count == self.next_input_count_to_log: |
| 106 | should_log = True |
| 107 | log_message = logging.info |
| 108 | self.update_next_input_count_to_log() |
| 109 | elif udf_logging.udf_log_level == logging.DEBUG: |
| 110 | should_log = True |
| 111 | log_message = logging.debug |
| 112 | |
| 113 | try: |
| 114 | try: |
| 115 | if should_log: |
| 116 | log_message("Row %s: Serialized Input: %s" % (self.input_count, input_str)) |