1 /***********************************************************************
2  * Copyright (c) 2013-2025 General Atomics Integrated Intelligence, Inc.
3  * All rights reserved. This program and the accompanying materials
4  * are made available under the terms of the Apache License, Version 2.0
5  * which accompanies this distribution and is available at
6  * https://www.apache.org/licenses/LICENSE-2.0
7  ***********************************************************************/
8 
9 package org.locationtech.geomesa.accumulo.tools.ingest
10 
11 import com.beust.jcommander.{Parameter, ParameterException, Parameters}
12 import com.typesafe.config.Config
13 import org.apache.hadoop.conf.Configuration
14 import org.apache.hadoop.fs.{FileContext, Path}
15 import org.apache.hadoop.mapreduce.Job
16 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat
17 import org.geotools.api.feature.simple.SimpleFeatureType
18 import org.locationtech.geomesa.accumulo.data.AccumuloDataStore
19 import org.locationtech.geomesa.accumulo.jobs.mapreduce.GeoMesaAccumuloFileOutputFormat
20 import org.locationtech.geomesa.accumulo.tools.AccumuloDataStoreCommand.AccumuloDistributedCommand
21 import org.locationtech.geomesa.accumulo.tools.AccumuloDataStoreParams
22 import org.locationtech.geomesa.accumulo.tools.ingest.AccumuloBulkIngestCommand.AccumuloBulkIngestParams
23 import org.locationtech.geomesa.index.conf.partition.TablePartition
24 import org.locationtech.geomesa.jobs.JobResult.JobSuccess
25 import org.locationtech.geomesa.jobs.mapreduce.ConverterCombineInputFormat
26 import org.locationtech.geomesa.jobs.{Awaitable, JobResult, StatusCallback}
27 import org.locationtech.geomesa.tools.DistributedRunParam.RunModes
28 import org.locationtech.geomesa.tools.DistributedRunParam.RunModes.RunMode
29 import org.locationtech.geomesa.tools._
30 import org.locationtech.geomesa.tools.ingest.IngestCommand.{IngestParams, Inputs}
31 import org.locationtech.geomesa.tools.ingest._
32 import org.locationtech.geomesa.tools.utils.{DistributedCopy, Prompt}
33 import org.locationtech.geomesa.utils.hadoop.HadoopDelegate
34 import org.locationtech.geomesa.utils.index.IndexMode
35 
36 import java.io.File
37 
38 class AccumuloBulkIngestCommand extends IngestCommand[AccumuloDataStore] with AccumuloDistributedCommand {
39 
40   override val name = "bulk-ingest"
41   override val params = new AccumuloBulkIngestParams()
42 
43   override protected def startIngest(
44       mode: RunMode,
45       ds: AccumuloDataStore,
46       sft: SimpleFeatureType,
47       converter: Config,
48       inputs: Inputs): Awaitable = {
49 
50     val maxSplitSize =
51       if (!params.combineInputs) { None } else { Option(params.maxSplitSize).map(_.intValue()).orElse(Some(0)) }
52 
53     // validate index param now that we have a datastore and the sft has been created
54     val index = params.loadIndex(ds, sft.getTypeName, IndexMode.Write).map(_.identifier)
55 
56     val partitions = TablePartition(ds, sft).map { tp =>
57       if (params.cqlFilter == null) {
58         throw new ParameterException(
59           s"Schema '${sft.getTypeName}' is a partitioned store. In order to bulk load, the '--cql' parameter " +
60               "must be used to specify the range of the input data set")
61       }
62       tp.partitions(params.cqlFilter).filter(_.nonEmpty).getOrElse {
63         throw new ParameterException(
64           s"Partition filter does not correspond to partition scheme ${tp.getClass.getSimpleName}. Please specify " +
65               "a valid filter using the '--cql' parameter")
66       }
67     }
68 
69 
70     mode match {
71       case RunModes.Local =>
72         throw new IllegalArgumentException("Bulk ingest must be run in distributed mode")
73 
74       case RunModes.Distributed =>
75         val conf = new Configuration()
76         // file output format doesn't let you write to an existing directory
77         val output = new Path(params.outputPath)
78         val context = FileContext.getFileContext(output.toUri, conf)
79         if (context.util.exists(output)) {
80           val warning = s"Output directory '$output' exists"
81           if (params.force) {
82             Command.user.warn(s"$warning - deleting it")
83           } else if (!Prompt.confirm(s"WARNING DATA MAY BE LOST: $warning. Delete it and continue (y/n)? ")) {
84             throw new ParameterException(s"Output directory '$output' exists")
85           }
86           context.delete(output, true)
87         }
88 
89         val tempPath = Option(params.tempPath).map { temp =>
90           val path = new Path(temp)
91           // get a new file context as this is likely to be a different filesystem (i.e. hdfs vs s3)
92           val tempContext = FileContext.getFileContext(path.toUri, conf)
93           val dir = tempContext.makeQualified(path)
94           if (tempContext.util.exists(dir)) {
95             Command.user.info(s"Deleting temp output path $dir")
96             tempContext.delete(dir, true)
97           }
98           dir
99         }
100 
101         Command.user.info(s"Running bulk ingestion in distributed ${if (params.combineInputs) "combine " else "" }mode")
102         new BulkConverterIngest(ds, connection, sft, converter, inputs.paths, output, tempPath, maxSplitSize,
103           index, partitions, libjarsFiles, libjarsPaths)
104 
105       case _ =>
106         throw new UnsupportedOperationException(s"Missing implementation for mode $mode")
107     }
108   }
109 
110   class BulkConverterIngest(
111       ds: AccumuloDataStore,
112       dsParams: Map[String, String],
113       sft: SimpleFeatureType,
114       converterConfig: Config,
115       paths: Seq[String],
116       output: Path,
117       tempOutput: Option[Path],
118       maxSplitSize: Option[Int],
119       index: Option[String],
120       partitions: Option[Seq[String]],
121       libjarsFiles: Seq[String],
122       libjarsPaths: Iterator[() => Seq[File]]
123     ) extends ConverterIngestJob(dsParams, sft, converterConfig, paths, libjarsFiles, libjarsPaths) {
124 
125     private var libjars: String = _
126 
127     override def configureJob(job: Job): Unit = {
128       super.configureJob(job)
129       val dest = tempOutput.getOrElse(output)
130       GeoMesaAccumuloFileOutputFormat.configure(job, ds, dsParams, sft, dest, index, partitions)
131       maxSplitSize.foreach { max =>
132         job.setInputFormatClass(classOf[ConverterCombineInputFormat])
133         if (max > 0) {
134           FileInputFormat.setMaxInputSplitSize(job, max.toLong)
135         }
136       }
137       this.libjars = job.getConfiguration.get("tmpjars")
138     }
139 
140     override def await(reporter: StatusCallback): JobResult = {
141       super.await(reporter).merge {
142         tempOutput.map { dir =>
143           reporter.reset()
144           val conf = new Configuration()
145           conf.set("tmpjars", this.libjars) // copy over out libjars so s3 apis are on the classpath
146           new DistributedCopy(conf).copy(Seq(dir), output, reporter) match {
147             case JobSuccess(message, counts) =>
148               Command.user.info(message)
149               JobSuccess("", counts)
150 
151             case j => j
152           }
153         }
154       }.merge {
155         if (params.skipImport) {
156           Command.user.info("Skipping import of RFiles into Accumulo")
157           Command.user.info(
158             "Files may be imported for each table through the Accumulo shell with the `importdirectory` command")
159         } else {
160           Command.user.info("Importing RFiles into Accumulo")
161           val tableOps = ds.client.tableOperations()
162           val filesPath = new Path(output, GeoMesaAccumuloFileOutputFormat.FilesPath)
163           val fc = FileContext.getFileContext(filesPath.toUri, new Configuration())
164           val files = fc.listLocatedStatus(filesPath)
165           while (files.hasNext) {
166             val file = files.next()
167             val path = file.getPath
168             val table = path.getName
169             if (file.isDirectory && HadoopDelegate.HiddenFileFilter.accept(path) && tableOps.exists(table)) {
170               Command.user.info(s"Importing $table")
171               tableOps.importDirectory(path.toString).to(table).load()
172             }
173           }
174         }
175         None
176       }
177     }
178   }
179 }
180 
181 object AccumuloBulkIngestCommand {
182   @Parameters(commandDescription = "Convert various file formats into bulk loaded Accumulo RFiles")
183   class AccumuloBulkIngestParams extends IngestParams with AccumuloDataStoreParams
184       with OutputPathParam with OptionalIndexParam with OptionalCqlFilterParam with TempPathParam {
185     @Parameter(names = Array("--skip-import"), description = "Generate the files but skip the bulk import into Accumulo")
186     var skipImport: Boolean = false
187   }
188 }