parameters: (self, load_node, pipeline)
docstring: 内部接口
docstring_summary: 内部接口
docstring_tokens: 内部接口
function: def transform_from_node(self, load_node, pipeline):
          """
          内部接口
          """
          from bigflow import schema
          if self.fields is None:
              raise ValueError('''columns is necessary，(1) columns(list)，
                  each item in columns is string, SchemaPCollection's element
                  is dict, (2) columns(int)，SchemaPCollection's element is tuple. eg.
                  columns=3 or columns=[(xx, int), (yy, str)] or columns=[xx, yy],
                  (3) columns(list), each item in columns is base type in [int, float, str]''')
  
          if isinstance(self.fields, tuple):
              self.fields = list(self.fields)
  
          fields_type = []
          ignore_overflow = self.ignore_overflow
          ignore_illegal_line = self.ignore_illegal_line
          if isinstance(self.fields, list):
              def get_fields_type(fields):
                  """内部函数"""
                  fields_type = []
                  for field in fields:
                      if isinstance(field, tuple):
                          if field[1] in [int, str, float]:
                              fields_type.append(field[1])
                          else:
                              raise ValueError('''columns is list(field name or data type),
                                               data type(int/str/float)''')
                      elif field in [int, str, float]:
                          fields_type.append(field)
                      elif isinstance(field, str):
                          fields_type.append(str)
                      else:
                          raise ValueError('''columns is list(field name or data type),
                                           data type(int/str/float)''')
                  return fields_type
              fields_type = get_fields_type(self.fields)
              ret = super(SchemaTextFile, self)\
                      .transform_from_node(load_node, pipeline)\
                      .flat_map(entity.SplitStringToTypes(self.sep,
                                                          fields_type,
                                                          ignore_overflow,
                                                          ignore_illegal_line),
                                                          serde=serde.of(tuple(fields_type)))
              if self.fields[0] in [int, float, str]:
                  return ret
              else:
                  ret = ret.apply(schema.tuple_to_dict, self.fields)
                  return ret
          elif isinstance(self.fields, int):
              from bigflow import schema_pcollection
              return schema_pcollection.SchemaPCollection(super(SchemaTextFile, self)
                  .transform_from_node(load_node, pipeline)\
                  .flat_map(entity.SplitStringToTypes(self.sep,
                                                      [str for _ in xrange(self.fields)],
                                                      True,
                                                      ignore_illegal_line),
                            serde=serde.of(tuple(serde.StrSerde() for index in xrange(self.fields)))))
          else:
              raise ValueError("columns is list(field name)，or int(row number)")
function_tokens: def transform_from_node ( self , load_node , pipeline ) : from bigflow import schema if self . fields is None : raise ValueError ( '''columns is necessary，(1) columns(list)，
                  each item in columns is string, SchemaPCollection's element
                  is dict, (2) columns(int)，SchemaPCollection's element is tuple. eg.
                  columns=3 or columns=[(xx, int), (yy, str)] or columns=[xx, yy],
                  (3) columns(list), each item in columns is base type in [int, float, str]''' ) if isinstance ( self . fields , tuple ) : self . fields = list ( self . fields ) fields_type = [ ] ignore_overflow = self . ignore_overflow ignore_illegal_line = self . ignore_illegal_line if isinstance ( self . fields , list ) : def get_fields_type ( fields ) : """内部函数""" fields_type = [ ] for field in fields : if isinstance ( field , tuple ) : if field [ 1 ] in [ int , str , float ] : fields_type . append ( field [ 1 ] ) else : raise ValueError ( '''columns is list(field name or data type),
                                               data type(int/str/float)''' ) elif field in [ int , str , float ] : fields_type . append ( field ) elif isinstance ( field , str ) : fields_type . append ( str ) else : raise ValueError ( '''columns is list(field name or data type),
                                           data type(int/str/float)''' ) return fields_type fields_type = get_fields_type ( self . fields ) ret = super ( SchemaTextFile , self ) . transform_from_node ( load_node , pipeline ) . flat_map ( entity . SplitStringToTypes ( self . sep , fields_type , ignore_overflow , ignore_illegal_line ) , serde = serde . of ( tuple ( fields_type ) ) ) if self . fields [ 0 ] in [ int , float , str ] : return ret else : ret = ret . apply ( schema . tuple_to_dict , self . fields ) return ret elif isinstance ( self . fields , int ) : from bigflow import schema_pcollection return schema_pcollection . SchemaPCollection ( super ( SchemaTextFile , self ) . transform_from_node ( load_node , pipeline ) . flat_map ( entity . SplitStringToTypes ( self . sep , [ str for _ in xrange ( self . fields ) ] , True , ignore_illegal_line ) , serde = serde . of ( tuple ( serde . StrSerde ( ) for index in xrange ( self . fields ) ) ) ) ) else : raise ValueError ( "columns is list(field name)，or int(row number)")