Romain Beaumont rom1504

This is about downloading http://the-eye.eu/eleuther_staging/cah/ which is a big dataset of image/text pairs filtered from common crawl

run get_links.sh ; this will produce a to_aria.txt file which contains all the urls to download and where to put them
run download.sh ; it will use aria2c to download files fast (takes about 1h)

Note if you only want one type of file, you may change this part grep 'csv\|txt\|pkl\|tfrecord'

This is a script to convert the csv of crawling at home to parquet

	[
	"t-shirt, size M",
	"flower dress, size M",
	"a t-shirt of an avocado",
	"a rainbow hat",
	"white snow covered mountain under blue sky during daytime",
	"aerial view of the beach during daytime",
	"aerial view of the beach at night",
	"double rainbow over a lake",
	"a beautiful sunset at a beach with a shell on the shore",

	from glob import glob
	import time
	import datetime

	def get_read_bytes():
	return sum([int(open(f"{p}/ports/1/hw_counters/rdma_read_bytes", "r").read().strip()) for p in glob("/sys/class/infiniband/*")])
	from os.path import expanduser
	home = expanduser("~")

	from multiprocessing.pool import ThreadPool

	import fsspec
	from tqdm import tqdm
	import sys

	m = int(sys.argv[1])
	t = int(sys.argv[2])

	path_s3 = "s3://laion5b/data/laion1B-nolang"

	from pyspark.sql import SparkSession
	import pyspark.sql.functions as F
	from urllib.parse import urlparse
	from pyspark.sql.functions import udf
	import os
	import fire

	def main(input_folder, output_folder):

	spark = SparkSession.builder.config("spark.driver.memory", "16G") .master("local[16]").appName('spark-stats').getOrCreate()

	from multiprocessing import Process, Queue
	import queue
	import time

	class StoppableProcess(Process):
	def __init__(self, lol):
	super().__init__()
	self.lol = lol
	self.q = Queue()

	# I advised to run this in an interactive environment (python shell, jupyter, ...) to understand well all the steps

	from pyspark.sql import SparkSession
	# Let's get tfrecord and rapids (rapids is not necessary, remove all mention if wanted)
	# wget https://search.maven.org/remotecontent?filepath=com/linkedin/sparktfrecord/spark-tfrecord_2.12/0.3.2/spark-tfrecord_2.12-0.3.2.jar -O tfrecord.jar
	# wget https://repo1.maven.org/maven2/com/nvidia/rapids-4-spark_2.12/21.06.0/rapids-4-spark_2.12-21.06.0.jar -O rapids.jar
	# wget https://repo1.maven.org/maven2/ai/rapids/cudf/21.06.1/cudf-21.06.1-cuda11.jar -O cudf.jar

	# creating the spark session with tfrecord plugin, rapids plugins, and some basic options, with local executor
	spark = SparkSession.builder.config("spark.jars", "tfrecord.jar,rapids.jar,cudf.jar").config("spark.plugins","com.nvidia.spark.SQLPlugin").config("spark.rapids.sql.incompatibleOps.enabled","true").config("spark.driver.memory", "16G") .master("local[16]").appName('spark-stats').getOrCreate()