func processUrl(url string, id string) {
	log.Println("Url", url, id)
	writtenNgrams := 0

	ctx, bucket, _ := prepareContext()

	targetRawFileName := getNgramFilenameFromUrl(url, true)
	targetFileName := getNgramFilenameFromUrl(url, false)
	targetObject := bucket.Object(targetFileName)
	_, err := targetObject.Attrs(ctx)
	if err == nil {
		// log.Println("Url", url, "already here:", targetFileName, "size", attrs.Size, id)
		return
	}

	targetRawObject := bucket.Object(targetRawFileName)
	_, err = targetRawObject.Attrs(ctx)
	if err == nil {
		// log.Println("Gzip already unzipped:", targetRawFileName, "size", attrs.Size, id)
	} else {
		response, err := http.Get(url)
		if err != nil {
			log.Fatal("Unable to Get URL", url, id, err)
		}
		defer response.Body.Close()

		gzReader, err := gzip.NewReader(response.Body)
		if err != nil {
			log.Fatal("Unable to open gzip", url, id, err)
		}
		defer gzReader.Close()

		gcRawWriter := targetRawObject.NewWriter(ctx)
		gcRawWriter.ContentType = "text/csv"

		log.Println("Extract gzip", url, "to", targetRawFileName, id)
		if _, err = io.Copy(gcRawWriter, gzReader); err != nil {
			log.Println("Unable to extract gzip", url, "to", targetRawFileName, id, err)
			return
		}
		if err := gcRawWriter.Close(); err != nil {
			log.Println("Unable to close raw", targetRawFileName, id, err)
			return
		}
		log.Println("Extract OK", targetRawFileName, id)
	}

	gcReader, err := targetRawObject.NewReader(ctx)
	if err != nil {
		log.Println("Unable to open raw", targetRawFileName, id, err)
		return
	}
	defer gcReader.Close()

	gcWriter := targetObject.NewWriter(ctx)
	gcWriter.ContentType = "text/csv"

	var isTheSame bool

	var bufferBuilder strings.Builder
	bufferLength := 0
	bufferCapacity := 100000
	var linesRead int64 = 0
	var linesWritten int = 0

	var ngram Ngram
	var previousNgram Ngram
	var tabSeparator = [...]byte{9}

	scanner := bufio.NewScanner(gcReader)
	// log.Println("Scanning", targetFileName, id)
	for scanner.Scan() {
		line := scanner.Bytes()
		linesRead++
		if linesRead%100000000 == 0 {
			log.Println("+100M lines", targetRawFileName, "total", linesRead, id)
		}

		// format: ngram TAB year TAB match_count TAB volume_count NEWLINE
		fragments := bytes.Split(line, tabSeparator[:])
		if len(fragments) >= 4 {
			isTheSame = prepareNgram(fragments[0], fragments[1], fragments[2], fragments[3], &ngram, &previousNgram)

			if !isTheSame && previousNgram.Text != "" {
				bufferBuilder.WriteString(previousNgram.Text)
				bufferBuilder.WriteString(";")
				bufferBuilder.WriteString(strconv.Itoa(previousNgram.Frequency))
				bufferBuilder.WriteString("\n")

				bufferLength++
				if bufferLength > bufferCapacity {
					linesWritten += bufferLength

					log.Println("Buffer flush", bufferLength, "total", linesWritten, "to", targetFileName, id)
					gcWriter.Write([]byte(bufferBuilder.String()))

					bufferBuilder.Reset()
					bufferLength = 0
				}

				writtenNgrams++
			}

			previousNgram.OriginalText = ngram.OriginalText
			previousNgram.Text = ngram.Text
			previousNgram.Frequency = ngram.Frequency
		} else {
			log.Println("WARNING: Found a line with not enough TABs:", string(line), id)
		}
	}

	if bufferLength > 0 {
		log.Println("Final buffer flush", bufferLength, "to", targetFileName, id)
		gcWriter.Write([]byte(bufferBuilder.String()))
		bufferBuilder.Reset()
	}

	if err := scanner.Err(); err != nil {
		log.Fatal(err)
	}

	if err := gcWriter.Close(); err != nil {
		log.Fatal("Unable to close", targetFileName, id, err)
	} else {
		// successfully writen => we can remove the raw file
		targetRawObject.Delete(ctx)
		log.Println("Deleted raw", targetRawFileName, id)
	}

	log.Println("======= Finished", url, "with", writtenNgrams, "ngrams written into", targetFileName, id)
}
// processUrl processes one URL and it's meant to run concurrently.
// it can handle files the size of tens of gigabytes within one hour timeout (tested)
processUrl processes one URL and it's meant to run concurrently.
it can handle files the size of tens of gigabytes within one hour timeout (tested)
// processUrl processes one URL and it ' s meant to run concurrently . // it can handle files the size of tens of gigabytes within one hour timeout ( tested )
func processUrl(url string, id string) {
	log.Println("Url", url, id)
	writtenNgrams := 0
	ctx, bucket, _ := prepareContext()
	targetRawFileName := getNgramFilenameFromUrl(url, true)
	targetFileName := getNgramFilenameFromUrl(url, false)
	targetObject := bucket.Object(targetFileName)
	_, err := targetObject.Attrs(ctx)
	if err == nil {
		return
	}
	targetRawObject := bucket.Object(targetRawFileName)
	_, err = targetRawObject.Attrs(ctx)
	if err == nil {
	} else {
		response, err := http.Get(url)
		if err != nil {
			log.Fatal("Unable to Get URL", url, id, err)
		}
		defer response.Body.Close()
		gzReader, err := gzip.NewReader(response.Body)
		if err != nil {
			log.Fatal("Unable to open gzip", url, id, err)
		}
		defer gzReader.Close()
		gcRawWriter := targetRawObject.NewWriter(ctx)
		gcRawWriter.ContentType = "text/csv"
		log.Println("Extract gzip", url, "to", targetRawFileName, id)
		if _, err = io.Copy(gcRawWriter, gzReader); err != nil {
			log.Println("Unable to extract gzip", url, "to", targetRawFileName, id, err)
			return
		}
		if err := gcRawWriter.Close(); err != nil {
			log.Println("Unable to close raw", targetRawFileName, id, err)
			return
		}
		log.Println("Extract OK", targetRawFileName, id)
	}
	gcReader, err := targetRawObject.NewReader(ctx)
	if err != nil {
		log.Println("Unable to open raw", targetRawFileName, id, err)
		return
	}
	defer gcReader.Close()
	gcWriter := targetObject.NewWriter(ctx)
	gcWriter.ContentType = "text/csv"
	var isTheSame bool
	var bufferBuilder strings.Builder
	bufferLength := 0
	bufferCapacity := 100000
	var linesRead int64 = 0
	var linesWritten int = 0
	var ngram Ngram
	var previousNgram Ngram
	var tabSeparator = [...]byte{9}
	scanner := bufio.NewScanner(gcReader)
	for scanner.Scan() {
		line := scanner.Bytes()
		linesRead++
		if linesRead%100000000 == 0 {
			log.Println("+100M lines", targetRawFileName, "total", linesRead, id)
		}
		fragments := bytes.Split(line, tabSeparator[:])
		if len(fragments) >= 4 {
			isTheSame = prepareNgram(fragments[0], fragments[1], fragments[2], fragments[3], &ngram, &previousNgram)
			if !isTheSame && previousNgram.Text != "" {
				bufferBuilder.WriteString(previousNgram.Text)
				bufferBuilder.WriteString(";")
				bufferBuilder.WriteString(strconv.Itoa(previousNgram.Frequency))
				bufferBuilder.WriteString("\n")
				bufferLength++
				if bufferLength > bufferCapacity {
					linesWritten += bufferLength
					log.Println("Buffer flush", bufferLength, "total", linesWritten, "to", targetFileName, id)
					gcWriter.Write([]byte(bufferBuilder.String()))
					bufferBuilder.Reset()
					bufferLength = 0
				}
				writtenNgrams++
			}
			previousNgram.OriginalText = ngram.OriginalText
			previousNgram.Text = ngram.Text
			previousNgram.Frequency = ngram.Frequency
		} else {
			log.Println("WARNING: Found a line with not enough TABs:", string(line), id)
		}
	}
	if bufferLength > 0 {
		log.Println("Final buffer flush", bufferLength, "to", targetFileName, id)
		gcWriter.Write([]byte(bufferBuilder.String()))
		bufferBuilder.Reset()
	}
	if err := scanner.Err(); err != nil {
		log.Fatal(err)
	}
	if err := gcWriter.Close(); err != nil {
		log.Fatal("Unable to close", targetFileName, id, err)
	} else {
		targetRawObject.Delete(ctx)
		log.Println("Deleted raw", targetRawFileName, id)
	}
	log.Println("======= Finished", url, "with", writtenNgrams, "ngrams written into", targetFileName, id)
}
func processUrl ( url string , id string ) { log . Println ( " " , url , id ) writtenNgrams := 0 ctx , bucket , _ := prepareContext ( ) targetRawFileName := getNgramFilenameFromUrl ( url , true ) targetFileName := getNgramFilenameFromUrl ( url , false ) targetObject := bucket . Object ( targetFileName ) _ , err := targetObject . Attrs ( ctx ) if err == nil { return } targetRawObject := bucket . Object ( targetRawFileName ) _ , err = targetRawObject . Attrs ( ctx ) if err == nil { } else { response , err := http . Get ( url ) if err != nil { log . Fatal ( " " , url , id , err ) } defer response . Body . Close ( ) gzReader , err := gzip . NewReader ( response . Body ) if err != nil { log . Fatal ( " " , url , id , err ) } defer gzReader . Close ( ) gcRawWriter := targetRawObject . NewWriter ( ctx ) gcRawWriter . ContentType = " " log . Println ( " " , url , " " , targetRawFileName , id ) if _ , err = io . Copy ( gcRawWriter , gzReader ) ; err != nil { log . Println ( " " , url , " " , targetRawFileName , id , err ) return } if err := gcRawWriter . Close ( ) ; err != nil { log . Println ( " " , targetRawFileName , id , err ) return } log . Println ( " " , targetRawFileName , id ) } gcReader , err := targetRawObject . NewReader ( ctx ) if err != nil { log . Println ( " " , targetRawFileName , id , err ) return } defer gcReader . Close ( ) gcWriter := targetObject . NewWriter ( ctx ) gcWriter . ContentType = " " var isTheSame bool var bufferBuilder strings . Builder bufferLength := 0 bufferCapacity := 100000 var linesRead int64 = 0 var linesWritten int = 0 var ngram Ngram var previousNgram Ngram var tabSeparator = [ ... ] byte { 9 } scanner := bufio . NewScanner ( gcReader ) for scanner . Scan ( ) { line := scanner . Bytes ( ) linesRead ++ if linesRead % 100000000 == 0 { log . Println ( " " , targetRawFileName , " " , linesRead , id ) } fragments := bytes . Split ( line , tabSeparator [ : ] ) if len ( fragments ) >= 4 { isTheSame = prepareNgram ( fragments [ 0 ] , fragments [ 1 ] , fragments [ 2 ] , fragments [ 3 ] , & ngram , & previousNgram ) if ! isTheSame && previousNgram . Text != " " { bufferBuilder . WriteString ( previousNgram . Text ) bufferBuilder . WriteString ( " " ) bufferBuilder . WriteString ( strconv . Itoa ( previousNgram . Frequency ) ) bufferBuilder . WriteString ( " \n " ) bufferLength ++ if bufferLength > bufferCapacity { linesWritten += bufferLength log . Println ( " " , bufferLength , " " , linesWritten , " " , targetFileName , id ) gcWriter . Write ( [ ] byte ( bufferBuilder . String ( ) ) ) bufferBuilder . Reset ( ) bufferLength = 0 } writtenNgrams ++ } previousNgram . OriginalText = ngram . OriginalText previousNgram . Text = ngram . Text previousNgram . Frequency = ngram . Frequency } else { log . Println ( " " , string ( line ) , id ) } } if bufferLength > 0 { log . Println ( " " , bufferLength , " " , targetFileName , id ) gcWriter . Write ( [ ] byte ( bufferBuilder . String ( ) ) ) bufferBuilder . Reset ( ) } if err := scanner . Err ( ) ; err != nil { log . Fatal ( err ) } if err := gcWriter . Close ( ) ; err != nil { log . Fatal ( " " , targetFileName , id , err ) } else { targetRawObject . Delete ( ctx ) log . Println ( " " , targetRawFileName , id ) } log . Println ( " " , url , " " , writtenNgrams , " " , targetFileName , id ) }
processUrl processes one URL and it's meant to run concurrently.
processUrl processes one URL and it ' s meant to run concurrently .
// log.Println("Url", url, "already here:", targetFileName, "size", attrs.Size, id) // log.Println("Gzip already unzipped:", targetRawFileName, "size", attrs.Size, id) // log.Println("Scanning", targetFileName, id) // format: ngram TAB year TAB match_count TAB volume_count NEWLINE // successfully writen => we can remove the raw file