@@ -1895,9 +1895,8 @@ def create_ingestion_dag(
18951895 use_sklearn : bool = True ,
18961896 mode : Mode = Mode .LOCAL ,
18971897 acn : Optional [str ] = None ,
1898- namespace : Optional [str ] = None
1898+ namespace : Optional [str ] = None ,
18991899 ) -> dag .DAG :
1900-
19011900 kwargs = {}
19021901 if mode == Mode .BATCH :
19031902 d = dag .DAG (
@@ -1908,7 +1907,7 @@ def create_ingestion_dag(
19081907 limit = 1 ,
19091908 retry_policy = "Always" ,
19101909 ),
1911- namespace = namespace
1910+ namespace = namespace ,
19121911 )
19131912 threads = 16
19141913 if acn :
@@ -1958,7 +1957,7 @@ def create_ingestion_dag(
19581957 name = "ingest" ,
19591958 resources = {"cpu" : str (threads ), "memory" : "16Gi" },
19601959 image_name = DEFAULT_IMG_NAME ,
1961- ** kwargs
1960+ ** kwargs ,
19621961 )
19631962 return d
19641963 elif index_type == "IVF_FLAT" :
@@ -1975,7 +1974,7 @@ def create_ingestion_dag(
19751974 name = "copy-centroids" ,
19761975 resources = {"cpu" : "1" , "memory" : "2Gi" },
19771976 image_name = DEFAULT_IMG_NAME ,
1978- ** kwargs
1977+ ** kwargs ,
19791978 )
19801979 else :
19811980 random_sample_nodes = []
@@ -2030,7 +2029,7 @@ def create_ingestion_dag(
20302029 name = "read-random-sample-" + str (idx ),
20312030 resources = {"cpu" : "2" , "memory" : "6Gi" },
20322031 image_name = DEFAULT_IMG_NAME ,
2033- ** kwargs
2032+ ** kwargs ,
20342033 )
20352034 )
20362035 num_sampled += num_to_sample
@@ -2059,7 +2058,7 @@ def create_ingestion_dag(
20592058 name = "kmeans" ,
20602059 resources = {"cpu" : "8" , "memory" : "32Gi" },
20612060 image_name = DEFAULT_IMG_NAME ,
2062- ** kwargs
2061+ ** kwargs ,
20632062 )
20642063
20652064 for random_sample_node in random_sample_nodes :
@@ -2088,7 +2087,7 @@ def create_ingestion_dag(
20882087 name = "init-centroids" ,
20892088 resources = {"cpu" : "1" , "memory" : "1Gi" },
20902089 image_name = DEFAULT_IMG_NAME ,
2091- ** kwargs
2090+ ** kwargs ,
20922091 )
20932092
20942093 for random_sample_node in random_sample_nodes :
@@ -2123,7 +2122,7 @@ def create_ingestion_dag(
21232122 name = "k-means-part-" + str (task_id ),
21242123 resources = {"cpu" : str (threads ), "memory" : "12Gi" },
21252124 image_name = DEFAULT_IMG_NAME ,
2126- ** kwargs
2125+ ** kwargs ,
21272126 )
21282127 )
21292128 task_id += 1
@@ -2136,7 +2135,7 @@ def create_ingestion_dag(
21362135 name = "update-centroids-" + str (i ),
21372136 resources = {"cpu" : "1" , "memory" : "8Gi" },
21382137 image_name = DEFAULT_IMG_NAME ,
2139- ** kwargs
2138+ ** kwargs ,
21402139 )
21412140 )
21422141 internal_centroids_node = submit (
@@ -2145,7 +2144,7 @@ def create_ingestion_dag(
21452144 name = "update-centroids" ,
21462145 resources = {"cpu" : "1" , "memory" : "8Gi" },
21472146 image_name = DEFAULT_IMG_NAME ,
2148- ** kwargs
2147+ ** kwargs ,
21492148 )
21502149 centroids_node = submit (
21512150 write_centroids ,
@@ -2159,7 +2158,7 @@ def create_ingestion_dag(
21592158 name = "write-centroids" ,
21602159 resources = {"cpu" : "1" , "memory" : "2Gi" },
21612160 image_name = DEFAULT_IMG_NAME ,
2162- ** kwargs
2161+ ** kwargs ,
21632162 )
21642163
21652164 compute_indexes_node = submit (
@@ -2172,7 +2171,7 @@ def create_ingestion_dag(
21722171 name = "compute-indexes" ,
21732172 resources = {"cpu" : "1" , "memory" : "2Gi" },
21742173 image_name = DEFAULT_IMG_NAME ,
2175- ** kwargs
2174+ ** kwargs ,
21762175 )
21772176
21782177 task_id = 0
@@ -2202,7 +2201,7 @@ def create_ingestion_dag(
22022201 name = "ingest-" + str (task_id ),
22032202 resources = {"cpu" : str (threads ), "memory" : "16Gi" },
22042203 image_name = DEFAULT_IMG_NAME ,
2205- ** kwargs
2204+ ** kwargs ,
22062205 )
22072206 ingest_node .depends_on (centroids_node )
22082207 compute_indexes_node .depends_on (ingest_node )
@@ -2222,7 +2221,7 @@ def create_ingestion_dag(
22222221 name = "ingest-" + str (task_id ),
22232222 resources = {"cpu" : str (threads ), "memory" : "16Gi" },
22242223 image_name = DEFAULT_IMG_NAME ,
2225- ** kwargs
2224+ ** kwargs ,
22262225 )
22272226 ingest_additions_node .depends_on (centroids_node )
22282227 compute_indexes_node .depends_on (ingest_additions_node )
@@ -2249,7 +2248,7 @@ def create_ingestion_dag(
22492248 name = "consolidate-partition-" + str (task_id ),
22502249 resources = {"cpu" : str (threads ), "memory" : "16Gi" },
22512250 image_name = DEFAULT_IMG_NAME ,
2252- ** kwargs
2251+ ** kwargs ,
22532252 )
22542253 consolidate_partition_node .depends_on (compute_indexes_node )
22552254 task_id += 1
0 commit comments