Skip to content

Change Feed demos

Jump to bottom
Khoa Dang edited this page Oct 17, 2017 · 1 revision

Below are the results of some change feed test runs in different scenarios using the CosmosDB Spark Connector.

CosmosDB Configurations

Partitioned Collection: 100,000 RUs, 50 partitions

Apache Spark Configurations

Azure HDI Clustger: 14 workers (4 cores, 28GB memory), 56 cores, 392GB memory

Change feed scenarios

Interval count

  • Set up the CosmosDB collection to have various insert load ranging from 10 to 1500 documents/second.
  • On the Spark cluster run a simple loop to process the change feed incrementally. In this case, we're only doing count.
import com.microsoft.azure.cosmosdb.spark.schema._
import com.microsoft.azure.cosmosdb.spark._
import com.microsoft.azure.cosmosdb.spark.config.Config
import org.codehaus.jackson.map.ObjectMapper
import org.slf4j.{Logger, LoggerFactory}

val log = LoggerFactory.getLogger("console")

val rollingChangeFeed = false
val startFromTheBeginning = false
val useNextToken = true

val configMap = Map(
"Endpoint" -> "COSMOSDB ENDPOINT",
"Masterkey" -> "COSMOSDB MASTER KEY",
"Database" -> "DATABASE NAME",
"Collection" -> "COLLECTION NAME",
"ReadChangeFeed" -> "true",
"ChangeFeedQueryName" -> s"change feed with rollingChangeFeed=$rollingChangeFeed, startFromtTheBeginning=$startFromTheBeginning, useNextToken=$useNextToken",
"ChangeFeedStartFromTheBeginning" -> startFromTheBeginning.toString,
"ChangeFeedUseNextToken" -> useNextToken.toString,
"RollingChangeFeed" -> rollingChangeFeed.toString,
"ChangeFeedCheckpointLocation" -> "./changefeedcheckpointlocation",
"SamplingRatio" -> "1.0")

val readConfig = Config(configMap)
val coll = spark.sparkContext.loadFromCosmosDB(readConfig)
var iteration = 1
val objectMapper = new ObjectMapper()

while (true){
    var result = coll.rdd.map(x => x.getId()).collect()
    var count = result.size
    val s = objectMapper.writeValueAsString(result)
    val resultString = s.substring(0, Math.min(s.length, 128))

    log.warn(s"change feed count is: $count, iteration = $iteration, result = $resultString")

    Thread.sleep(3000)
    iteration = iteration + 1
}

Observations

We've noticed that the change feed documents were received correctly for all configurations of insert load. In case of node failures, the connector were able to resume the change feed since the last checkpoint. Below are the highlights from the test run logs.

17/08/18 21:16:31 WARN console: change feed count is: 0, iteration = 1, result = []
17/08/18 21:16:35 WARN console: change feed count is: 0, iteration = 2, result = []
17/08/18 21:16:39 WARN console: change feed count is: 0, iteration = 3, result = []
17/08/18 21:16:43 WARN console: change feed count is: 0, iteration = 4, result = []
# Writing 100 documents/s in 20s
17/08/18 21:16:47 WARN console: change feed count is: 249, iteration = 5, result = ["user1099968472986350811","user1104996300299917057","user1135913194976225383","user1216970076899119993","user124788697157542831..
17/08/18 21:16:51 WARN console: change feed count is: 380, iteration = 6, result = ["user1016683566310592195","user1026397575643426122","user1104654655005618492","user1117386782116996542","user112263292464840467..
17/08/18 21:16:55 WARN console: change feed count is: 391, iteration = 7, result = ["user1021711393624158441","user1109682482319184738","user1117946742629136990","user1122072964136264223","user112766075196197091..
17/08/18 21:16:59 WARN console: change feed count is: 394, iteration = 8, result = ["user1009539227024920839","user1099408512474210363","user1110460758049167069","user1110802403343465634","user111682682160485609..
17/08/18 21:17:03 WARN console: change feed count is: 403, iteration = 9, result = ["user1020591472599877545","user1021151433112017993","user1105214615517758940","user1110242442831325186","user111660850638701421..
17/08/18 21:17:07 WARN console: change feed count is: 183, iteration = 10, result = ["user1104436339787776609","user1111020718561307517","user1241862553314037859","user1248446932087568767","user132592573571977880..
17/08/18 21:17:11 WARN console: change feed count is: 0, iteration = 11, result = []
17/08/18 21:17:15 WARN console: change feed count is: 0, iteration = 12, result = []
...
# Writing 100k documents (~1500 documents/s)
17/08/18 21:17:50 WARN console: change feed count is: 2752, iteration = 21, result = ["user1000726823498525925","user1001286784010666373","user1004293084493512710","user1004853045005653158","user100541300551779360..
17/08/18 21:17:54 WARN console: change feed count is: 5897, iteration = 22, result = ["user1005972966029934054","user1006191281247775937","user1006532926542074502","user1006751241759916385","user100731120227205683..
17/08/18 21:17:57 WARN console: change feed count is: 5447, iteration = 23, result = ["user1002625020252789152","user10046604894068544","user100475850855498516","user1005640680888162","user101148895975152061","use..
17/08/18 21:18:01 WARN console: change feed count is: 6178, iteration = 24, result = ["user1000166862986385477","user100028974950871165","user1000839908106039022","user1001058223323880905","user1001399868618179470..
17/08/18 21:18:05 WARN console: change feed count is: 6275, iteration = 25, result = ["user1000053778378872380","user1000272093596714263","user1000613738891012828","user1000832054108854711","user100251193564527605..
17/08/18 21:18:09 WARN console: change feed count is: 4309, iteration = 26, result = ["user100922726760125867","user1011779069073482631","user1019036492966667084","user1022831314648439337","user1022936545258768123..
17/08/18 21:18:13 WARN console: change feed count is: 5631, iteration = 27, result = ["user1000945138716367808","user1004626875790626964","user1005526090125306703","user1005744405343148586","user100675909575710069..
17/08/18 21:18:17 WARN console: change feed count is: 6755, iteration = 28, result = ["user100142059558384262","user1002958811549903406","user1005186836302767412","user1005746796814907860","user1005965112032749743..
17/08/18 21:18:21 WARN console: change feed count is: 6220, iteration = 29, result = ["user1000952992713552119","user1001171307931394002","user100362766247985419","user1005299920910280509","user1005518236128122392..
17/08/18 21:18:26 WARN console: change feed count is: 6979, iteration = 30, result = ["user1000279947593898574","user1000724432026766651","user1001284392538907099","user1006646011149587599","user100686432636742948..
17/08/18 21:18:30 WARN console: change feed count is: 6933, iteration = 31, result = ["user1000393032201411671","user1000498262811740457","user1000506116808924768","user1000611347419253554","user100284572694239030..
17/08/18 21:18:34 WARN console: change feed count is: 5343, iteration = 32, result = ["user1002372952224621199","user1005859881422420957","user1012112860370596885","user1012305276775297009","user101286523728743745..
17/08/18 21:18:39 WARN console: change feed count is: 6783, iteration = 33, result = ["user1000133110176059407","user1000474755470357972","user100062727761197235","user1000693070688199855","user1001253031200340303..
17/08/18 21:18:43 WARN console: change feed count is: 8157, iteration = 34, result = ["user1000532015622066527","user1000750330839908410","user1001651936646347423","user1001870251864189306","user100299017288847020..
17/08/18 21:18:47 WARN console: change feed count is: 6107, iteration = 35, result = ["user1005047852882112556","user1005266168099954439","user1005492337314980633","user1006052297827121081","user100773217936354242..
17/08/18 21:18:51 WARN console: change feed count is: 5764, iteration = 36, result = ["user1000305846407040333","user1000524161624882216","user1001310291352048858","user1003550133400610650","user100448789236997210..
17/08/18 21:18:55 WARN console: change feed count is: 4422, iteration = 37, result = ["user1001091976134206975","user1002764003673444008","user100449952042356757","user1007344955082382903","user1012362536927005564..
17/08/18 21:18:59 WARN console: change feed count is: 48, iteration = 38, result = ["user1147047163820128390","user1181216199487191344","user1538398494424988179","user1593552100217683649","user159784176826270348..
17/08/18 21:19:02 WARN console: change feed count is: 0, iteration = 39, result = []
17/08/18 21:19:06 WARN console: change feed count is: 0, iteration = 40, result = []
...
# Writing ~1000 documents/s
17/08/18 21:21:24 WARN console: change feed count is: 688, iteration = 68, result = ["user1004293084493512710","user1005413005517793606","user1010099187537061287","user1015345330068469416","user109640221199136402
17/08/18 21:21:34 WARN console: change feed count is: 4322, iteration = 69, result = ["user1000726823498525925","user1001286784010666373","user1004853045005653158","user1005631320735635489","user100843112329633772
17/08/18 21:21:38 WARN console: change feed count is: 7461, iteration = 70, result = ["user10046604894068544","user100475850855498516","user1005640680888162","user1005972966029934054","user1006191281247775937","us
17/08/18 21:21:43 WARN console: change feed count is: 3062, iteration = 71, result = ["user100028974950871165","user1001505099228508256","user1002625020252789152","user1011219108561342183","user1011437423779184066
17/08/18 21:21:51 WARN console: change feed count is: 3632, iteration = 72, result = ["user1000166862986385477","user1000839908106039022","user1001058223323880905","user1001399868618179470","user100161818383602135
17/08/18 21:21:55 WARN console: change feed count is: 6343, iteration = 73, result = ["user1000053778378872380","user1000272093596714263","user1000613738891012828","user1000832054108854711","user100251193564527605
17/08/18 21:21:59 WARN console: change feed count is: 3074, iteration = 74, result = ["user101035811367638964","user10159689501581641","user1022936545258768123","user10272774109094738","user103150751512989873","us
17/08/18 21:22:02 WARN console: change feed count is: 3107, iteration = 75, result = ["user100922726760125867","user1011779069073482631","user1011997384291324514","user1018250363239500442","user1019036492966667084
17/08/18 21:22:06 WARN console: change feed count is: 2972, iteration = 76, result = ["user1000945138716367808","user1005526090125306703","user1005744405343148586","user1006977410974942579","user100719811766454373
17/08/18 21:22:10 WARN console: change feed count is: 2892, iteration = 77, result = ["user100142059558384262","user1004626875790626964","user1005746796814907860","user1006759095757100696","user1013903435042772052
17/08/18 21:22:14 WARN console: change feed count is: 3031, iteration = 78, result = ["user1005186836302767412","user1005965112032749743","user1006199135244960248","user1006306757327048308","user100641745046280213
17/08/18 21:22:17 WARN console: change feed count is: 3082, iteration = 79, result = ["user1002958811549903406","user1011771215076298320","user1012891136100579216","user1018697239144127793","user102148522440913224
17/08/18 21:22:21 WARN console: change feed count is: 3024, iteration = 80, result = ["user1000952992713552119","user1006638157152403288","user1011552899858456437","user101261980582665158","user1015119160853443222
17/08/18 21:22:25 WARN console: change feed count is: 3018, iteration = 81, result = ["user1001171307931394002","user100362766247985419","user1005299920910280509","user1005518236128122392","user1006086050637447151
17/08/18 21:22:29 WARN console: change feed count is: 2975, iteration = 82, result = ["user1006864326367429482","user1007650456094596124","user100815104678037807","user1015232245460956319","user1015458414675982513
17/08/18 21:22:33 WARN console: change feed count is: 3119, iteration = 83, result = ["user1000498262811740457","user1000724432026766651","user1001284392538907099","user1013230389923118507","user101375065190178255
17/08/18 21:22:37 WARN console: change feed count is: 3108, iteration = 84, result = ["user1000279947593898574","user1002845726942390309","user1009878480847460130","user1011999775763083788","user101222594497810998
17/08/18 21:22:40 WARN console: change feed count is: 3173, iteration = 85, result = ["user1000393032201411671","user1000506116808924768","user1000611347419253554","user1005073751695254315","user100588935463011613
17/08/18 21:22:45 WARN console: change feed count is: 3006, iteration = 86, result = ["user1005859881422420957","user1006078196640262840","user1012112860370596885","user1018584154536614696","user102143915096222074
17/08/18 21:22:49 WARN console: change feed count is: 3623, iteration = 87, result = ["user1002372952224621199","user1012305276775297009","user1012865237287437457","user1013425197799577905","user102335752235025371
17/08/18 21:22:53 WARN console: change feed count is: 2862, iteration = 88, result = ["user1000133110176059407","user1000474755470357972","user1000693070688199855","user1001253031200340303","user100181299171248075
17/08/18 21:22:57 WARN console: change feed count is: 3110, iteration = 89, result = ["user100062727761197235","user100281042979039118","user1008123230812365234","user100841003491179566","user1009287158996752886",
17/08/18 21:23:01 WARN console: change feed count is: 3148, iteration = 90, result = ["user1007003309788084338","user1016717319120918265","user1022636506771979939","user1022854821989821822","user102375642779626083
17/08/18 21:23:05 WARN console: change feed count is: 2197, iteration = 91, result = ["user1000750330839908410","user1001651936646347423","user1005665073545961559","user1011584261197023233","user101180257641486511
17/08/18 21:23:09 WARN console: change feed count is: 3007, iteration = 92, result = ["user1000532015622066527","user1001870251864189306","user1002990172888470202","user1006784994570242455","user100694604963637578
17/08/18 21:23:12 WARN console: change feed count is: 2966, iteration = 93, result = ["user1007732179363542425","user1009904379660601889","user10125936691255571","user1018784424938499131","user1027152344032025968"
17/08/18 21:23:16 WARN console: change feed count is: 2979, iteration = 94, result = ["user1005492337314980633","user1006052297827121081","user1010464340172742337","user1015540137944928814","user101632626767209545
17/08/18 21:23:20 WARN console: change feed count is: 3160, iteration = 95, result = ["user1003550133400610650","user1004487892369972108","user1005047852882112556","user1005266168099954439","user100527402209713875
17/08/18 21:23:24 WARN console: change feed count is: 3041, iteration = 96, result = ["user1000305846407040333","user1000524161624882216","user1001310291352048858","user1005105113033821111","user100834940002739142
17/08/18 21:23:28 WARN console: change feed count is: 3002, iteration = 97, result = ["user1001091976134206975","user1002764003673444008","user1007344955082382903","user1012144221709163681","user101236253692700556
17/08/18 21:23:31 WARN console: change feed count is: 1816, iteration = 98, result = ["user100449952042356757","user1020673195868823846","user1021233156380964294","user102898683484821920","user1044568261690912425"
17/08/18 21:23:35 WARN console: change feed count is: 0, iteration = 99, result = []
...
# Writing 100k documents with node failures
17/08/18 21:23:47 WARN console: change feed count is: 3541, iteration = 102, result = ["user1000726823498525925","user1001286784010666373","user1004293084493512710","user1004853045005653158","user100541300551779360
17/08/18 21:23:50 WARN console: change feed count is: 5875, iteration = 103, result = ["user1005640680888162","user1005972966029934054","user1006191281247775937","user1006532926542074502","user1006751241759916385",
17/08/18 21:23:54 WARN console: change feed count is: 6128, iteration = 104, result = ["user100028974950871165","user1001505099228508256","user1002625020252789152","user10046604894068544","user100475850855498516","
17/08/18 21:23:58 WARN console: change feed count is: 5119, iteration = 105, result = ["user1000166862986385477","user1000839908106039022","user1001058223323880905","user1001399868618179470","user100161818383602135
17/08/18 21:24:02 WARN console: change feed count is: 6121, iteration = 106, result = ["user1000053778378872380","user1000272093596714263","user1000613738891012828","user1000832054108854711","user100251193564527605
17/08/18 21:24:06 WARN console: change feed count is: 6082, iteration = 107, result = ["user1000945138716367808","user1005526090125306703","user1005744405343148586","user1007198117664543736","user100922726760125867
17/08/18 21:24:08 ERROR LiveListenerBus: SparkListenerBus has already stopped! Dropping event SparkListenerExecutorMetricsUpdate(1,WrappedArray())
...
17/08/18 21:25:15 WARN console: change feed count is: 67134, iteration = 1, result = ["user1000133110176059407","user1000279947593898574","user1000305846407040333","user1000393032201411671","user100047475547035797
17/08/18 21:25:19 WARN console: change feed count is: 0, iteration = 2, result = []
17/08/18 21:25:23 WARN console: change feed count is: 0, iteration = 3, result = []
17/08/18 21:25:27 WARN console: change feed count is: 0, iteration = 4, result = []
Clone this wiki locally