def _get_input_schema(self):
        """Arbitrarily picks an object in input and reads the Avro schema from it."""
        assert avro, 'avro module required'

        input_target = flatten(self.input())[0]
        input_fs = input_target.fs if hasattr(input_target, 'fs') else GCSClient()
        input_uri = self.source_uris()[0]
        if '*' in input_uri:
            file_uris = list(input_fs.list_wildcard(input_uri))
            if file_uris:
                input_uri = file_uris[0]
            else:
                raise RuntimeError('No match for ' + input_uri)

        schema = []
        exception_reading_schema = []

        def read_schema(fp):
            # fp contains the file part downloaded thus far. We rely on that the DataFileReader
            # initializes itself fine as soon as the file header with schema is downloaded, without
            # requiring the remainder of the file...
            try:
                reader = avro.datafile.DataFileReader(fp, avro.io.DatumReader())
                schema[:] = [reader.datum_reader.writers_schema]
            except Exception as e:
                # Save but assume benign unless schema reading ultimately fails. The benign
                # exception in case of insufficiently big downloaded file part seems to be:
                # TypeError('ord() expected a character, but string of length 0 found',).
                exception_reading_schema[:] = [e]
                return False
            return True

        input_fs.download(input_uri, 64 * 1024, read_schema).close()
        if not schema:
            raise exception_reading_schema[0]
        return schema[0]
def _get_input_schema ( self ) : assert avro , 'avro module required' input_target = flatten ( self . input ( ) ) [ 0 ] input_fs = input_target . fs if hasattr ( input_target , 'fs' ) else GCSClient ( ) input_uri = self . source_uris ( ) [ 0 ] if '*' in input_uri : file_uris = list ( input_fs . list_wildcard ( input_uri ) ) if file_uris : input_uri = file_uris [ 0 ] else : raise RuntimeError ( 'No match for ' + input_uri ) schema = [ ] exception_reading_schema = [ ] def read_schema ( fp ) : try : reader = avro . datafile . DataFileReader ( fp , avro . io . DatumReader ( ) ) schema [ : ] = [ reader . datum_reader . writers_schema ] except Exception as e : exception_reading_schema [ : ] = [ e ] return False return True input_fs . download ( input_uri , 64 * 1024 , read_schema ) . close ( ) if not schema : raise exception_reading_schema [ 0 ] return schema [ 0 ]
Arbitrarily picks an object in input and reads the Avro schema from it.