code: func (m *Main) Run() error {
  	src, err := NewSource(
  		OptSrcBucket(m.Bucket),
  		OptSrcPrefix(m.Prefix),
  		OptSrcRegion(m.Region),
  		OptSrcBufSize(1000),
  		OptSrcSubjectAt(m.SubjectAt),
  	)
  	if err != nil {
  		return errors.Wrap(err, "getting s3 source")
  	}
  
  	parser := pdk.NewDefaultGenericParser()
  	if len(m.SubjectPath) == 0 && m.SubjectAt == "" {
  		parser.Subjecter = pdk.BlankSubjecter{}
  	} 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
  
  	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))
  		log.Fatal(errors.Wrap(err, "starting mapping proxy"))
  	}()
  	return errors.Wrap(ingester.Run(), "running ingester")
  }
docs: // Run runs the ingester.