code: func (m *Main) Run() error {
  	src, err := NewSource(
  		OptSrcPath(m.Path),
  		OptSrcSubjectAt(m.SubjectAt),
  	)
  	if err != nil {
  		return errors.Wrap(err, "getting file source")
  	}
  
  	translateColumns := true
  	parser := pdk.NewDefaultGenericParser()
  	if len(m.SubjectPath) == 0 && m.SubjectAt == "" {
  		parser.Subjecter = pdk.BlankSubjecter{}
  		translateColumns = false
  	} else if len(m.SubjectPath) == 0 && m.SubjectAt != "" {
  		m.SubjectPath = []string{m.SubjectAt}
  		parser.EntitySubjecter = pdk.SubjectPath(m.SubjectPath)
  	} else {
  		parser.EntitySubjecter = pdk.SubjectPath(m.SubjectPath)
  	}
  
  	mapper := pdk.NewCollapsingMapper()
  	mapper.Framer = &m.Framer
  	if translateColumns {
  		mapper.ColTranslator = pdk.NewMapFieldTranslator()
  	}
  
  	indexer, err := pdk.SetupPilosa(m.PilosaHosts, m.Index, nil, m.BatchSize)
  	if err != nil {
  		return errors.Wrap(err, "setting up Pilosa")
  	}
  	ingester := pdk.NewIngester(src, parser, mapper, indexer)
  
  	go func() {
  		err = pdk.StartMappingProxy(m.Proxy, pdk.NewPilosaForwarder(m.PilosaHosts[0], mapper.Translator, mapper.ColTranslator))
  		log.Fatal(errors.Wrap(err, "starting mapping proxy"))
  	}()
  	return errors.Wrap(ingester.Run(), "running ingester")
  }
docs: // Run runs the ingester.