Importing Data

using CsvHelper;
using CsvHelper.Configuration;
using System.Globalization;

public async Task importCSV(string filename)
{
    using (var reader = new StreamReader(filename))
    using (var csv = new CsvReader(reader, CultureInfo.InvariantCulture))
    {
        var records = csv.GetRecords<dynamic>();
        foreach (dynamic record in records) {
            await upsertDocument(record);
        }
    }
}

public async Task importTSV(string filename)
{
    using (var reader = new StreamReader("import.tsv"))
    using (var tsv = new CsvReader(reader,
        new CsvConfiguration(CultureInfo.InvariantCulture)
            { Delimiter = "\t" }))
    {
        var records = tsv.GetRecords<dynamic>();
        foreach (dynamic record in records) {
            await upsertDocument(record);
        }
    }
}

import com.opencsv.*;

public void importCSV() {
  try (CSVReaderHeaderAware csv = new CSVReaderHeaderAware(
      new FileReader("modules/howtos/examples/import.csv"))) {
          
    Map<String, String> row;
    while ((row = csv.readMap()) != null) {
      upsertRow(row);
    }
  }
  catch (java.io.FileNotFoundException e) {
    System.out.println("handle FileNotFoundException...");
  }
  catch (java.io.IOException e) {
    System.out.println("handle IOException...");
  }
  catch (com.opencsv.exceptions.CsvValidationException e) {
    System.out.println("handle CsvValidationException...");
  }
}

public void importTSV() {
  CSVParser parser =
    new CSVParserBuilder()
    .withSeparator('\t')
    .withIgnoreQuotations(true)
    .build();
  
  try (CSVReaderHeaderAware tsv =
      new CSVReaderHeaderAwareBuilder(
        new FileReader("modules/howtos/examples/import.tsv"))
      .withCSVParser(parser)
      .build()) {

    Map<String, String> row;
    while ((row = tsv.readMap()) != null) {
      upsertRow(row);
    }
  }
  // ...
}

public void importCSV_batch() {
  
  Flux<Map<String,String>> rows = Flux.generate(
    
    () -> new CSVReaderHeaderAware(
      new FileReader("modules/howtos/examples/import.csv")),
    
    (state, sink) -> {
      try {
        Map<String,String> row = state.readMap();
        if (row == null) { sink.complete(); }
        else { sink.next(row); }
        return state;
      }
      catch (Exception e) { throw new RuntimeException(e); }
    },
    state -> {
      try { state.close(); }
      catch (Exception e) { throw new RuntimeException(e); }
    }
  );

  Flux<MutationResult> results = 
    rows
    .map(row -> preprocess(row))
    .flatMap(doc -> reactiveCollection.upsert(doc.getId(), doc.getContent()))
    .doOnNext(System.out::println);

  results.blockLast(Duration.ofSeconds(60));
}

public void importTSV_batch() {

  Flux<Map<String,String>> rows = Flux.generate(
    
    () -> {
      CSVParser parser =
        new CSVParserBuilder()
        .withSeparator('\t')
        .withIgnoreQuotations(true)
        .build();
      return
        new CSVReaderHeaderAwareBuilder(
          new FileReader("modules/howtos/examples/import.tsv"))
        .withCSVParser(parser)
        .build();
    },
    
    // ...
}

const { parse: csvParser } = require('csv-parse');

const csvStream = (filename) =>
  fs.createReadStream(filename)
  .pipe(
      csvParser({columns: true}))

const tsvStream = (filename) =>
  fs.createReadStream(filename)
    .pipe(
      csvParser({
        columns: true,
        delimiter: '\t'}))

import csv

def csv_import(filename):
    with open(filename, newline='') as csvfile:
        reader = csv.DictReader(csvfile)
        for row in reader:
            upsert(row)

def tsv_import(filename):
    with open(filename, newline='') as tsvfile:
        reader = csv.DictReader(tsvfile, delimiter='\t')
        for row in reader:
            upsert(row)

using Newtonsoft.Json;
using Newtonsoft.Json.Linq;

public async Task importJSON(string filename)
{
    using (var reader = new StreamReader("import.json"))
    {
        var jsonReader = new JsonTextReader(reader);
        JArray arr = (JArray)JToken.ReadFrom(jsonReader);
        
        foreach (JObject record in arr)
        {
            await upsertDocument(record);
        }
    }
}

public async Task importJSONL(string filename)
{
    using (var reader = new StreamReader("import.jsonl"))
    {
        var jsonlReader = new JsonTextReader(reader)
        {
            SupportMultipleContent = true
        };
        while (jsonlReader.Read())
        {
            var record = (JObject)JToken.ReadFrom(jsonlReader);
            await upsertDocument(record);
        }
    }
}

import com.couchbase.client.java.json.*;

public void importJSON() {
  try {
    String content  = new String(
      Files.readAllBytes( // read whole document into memory
        Paths.get("modules/howtos/examples/import.json")),
      StandardCharsets.UTF_8);
    
    for (Object row: JsonArray.fromJson(content)) {
      JsonObject json = ((JsonObject) row);
      upsertRow(json.toMap());
    }
  }
  catch (java.io.FileNotFoundException e) {
    System.out.println("handle FileNotFoundException...");
  }
  catch (java.io.IOException e) {
    System.out.println("handle IOException...");
  }
}

public void importJSON_batch() {

  try {
    String content  = new String(
      Files.readAllBytes( // read whole document into memory
        Paths.get("modules/howtos/examples/import.json")),
      StandardCharsets.UTF_8);
    
    Flux<MutationResult> results = 
      Flux.fromIterable(JsonArray.fromJson(content))
        .map(row -> ((JsonObject) row).toMap())
        .map(map -> preprocess(map))
        .flatMap(doc -> reactiveCollection.upsert(doc.getId(), doc.getContent()))
        .doOnNext(System.out::println);

    results.blockLast(Duration.ofSeconds(60));
  }
  // ...
}

public void importJSONL() {
  try (BufferedReader br =
        new BufferedReader(
          new FileReader("modules/howtos/examples/import.jsonl"))) {
            
      String line;
      while ((line = br.readLine()) != null) {
        Map<String,Object> row =
          JsonObject.fromJson(line).toMap();
          
        upsertRow(row);
    }
  }
  // ...
}

public void importJSONL_batch() {

  Flux<String> lines = Flux.using(
    () -> Files.lines(Paths.get("modules/howtos/examples/import.jsonl")),
    Flux::fromStream,
    BaseStream::close);

  Flux<MutationResult> results =
    lines
        .map(line -> JsonObject.fromJson(line).toMap())
        .map(map -> preprocess(map))
        .flatMap(doc -> reactiveCollection.upsert(doc.getId(), doc.getContent()))
        .doOnNext(System.out::println);

  results.blockLast(Duration.ofSeconds(60));
}

const stream = require('stream'); 

// for JSON
const StreamArray = require('stream-json/streamers/StreamArray')

// for JsonL
const {parser: jsonlParser} = require('stream-json/jsonl/Parser');

const map = (f) =>
  new stream.Transform({
    objectMode: true,
    transform: (obj, _, next) => next(null, f(obj))
  })
  
const jsonStream = (filename) =>
  fs.createReadStream(filename)
    .pipe(StreamArray.withParser())
    .pipe(map(obj => obj.value))

const jsonlStream = (filename) =>
  fs.createReadStream(filename)
    .pipe(jsonlParser())
    .pipe(map(obj => obj.value))

import json

def json_import(filename):
    with open(filename) as jsonfile:
        data = json.load(jsonfile)
        for row in data:
            upsert(row)

def jsonl_import(filename):
    with open(filename) as jsonlfile:
        for line in jsonlfile:
            row = json.loads(line)
            upsert(row)